"""Team model functions — team CRUD, invites, member management."""
import json
import secrets
from datetime import UTC, datetime, timedelta, timezone
from app.crypto import (
decrypt_entity,
decrypt_user_value,
encrypt_entity,
encrypt_user_submission,
encrypt_user_value,
encrypt_value,
hash_value,
is_encrypted,
key_is_configured,
try_decrypt,
try_decrypt_entity,
try_decrypt_user_submission,
try_decrypt_user_value,
)
from app.db import get_db
# Import helpers
from app.helpers import (
_decrypt_team_member,
_table_exists,
)
def create_team(name, owner_id):
"""Create a new team. Owner becomes the team owner."""
conn = None
try:
conn = get_db()
conn.execute(
"INSERT INTO teams (name, owner_id) VALUES (?, ?)",
(name, owner_id),
)
conn.commit()
team = conn.execute("SELECT * FROM teams WHERE id = ?", (conn.lastrowid,)).fetchone()
# Add owner as first member with 'owner' role
conn.execute(
"INSERT INTO team_members (team_id, user_id, role, status, joined_at) VALUES (?, ?, 'owner', 'active', ?)",
(conn.lastrowid, owner_id, datetime.now(UTC).isoformat()),
)
conn.commit()
return dict(team)
finally:
if conn:
conn.close()
def get_team(team_id):
"""Get team by ID."""
conn = None
try:
conn = get_db()
team = conn.execute("SELECT * FROM teams WHERE id = ?", (team_id,)).fetchone()
finally:
if conn:
conn.close()
return dict(team) if team else None
def get_user_teams(user_id):
"""Get all teams a user is a member of."""
conn = None
try:
conn = get_db()
rows = conn.execute(
"""SELECT t.*, tm.role, tm.status, tm.joined_at
FROM team_members tm
JOIN teams t ON t.id = tm.team_id
WHERE tm.user_id = ? AND tm.status = 'active'
ORDER BY t.created_at DESC""",
(user_id,),
).fetchall()
finally:
if conn:
conn.close()
return [dict(r) for r in rows]
def get_team_members(team_id):
"""Get all members of a team (active + pending invites)."""
conn = None
try:
conn = get_db()
rows = conn.execute(
"""SELECT tm.*, u.email_encrypted, u.name_encrypted
FROM team_members tm
LEFT JOIN users u ON u.id = tm.user_id
WHERE tm.team_id = ?
ORDER BY tm.role = 'owner' DESC, tm.created_at ASC""",
(team_id,),
).fetchall()
finally:
if conn:
conn.close()
return [_decrypt_team_member(dict(r)) for r in rows]
def invite_to_team(team_id, inviter_user_id, email, role="member"):
"""Invite someone to a team via email. Returns invite token."""
conn = None
try:
conn = get_db()
# Check if already invited or member — use hash for email lookup
email_hash = hash_value(email)
existing = conn.execute(
"SELECT * FROM team_members WHERE team_id = ? AND (invited_email_hash = ? OR (user_id IS NOT NULL AND (SELECT email FROM users WHERE users.id = user_id) = ?))",
(team_id, email_hash, email),
).fetchone()
if existing:
return {"error": "Already a member or invited"}
# Check member limit for team tier
team = conn.execute("SELECT * FROM teams WHERE id = ?", (team_id,)).fetchone()
if not team:
return {"error": "Team not found"}
active_count = conn.execute(
"SELECT count(*) as cnt FROM team_members WHERE team_id = ? AND status = 'active'",
(team_id,),
).fetchone()["cnt"]
if team["tier"] == "teams" and active_count >= 5:
return {"error": "Team tier limited to 5 members. Upgrade to Teams Plus for unlimited."}
token = secrets.token_urlsafe(32)
expires_at = (datetime.now(UTC) + timedelta(days=INVITE_TOKEN_EXPIRY_DAYS)).isoformat()
# Insert with hash for lookup, encrypted email in the encrypted column
conn.execute(
"INSERT INTO team_members (team_id, role, invited_email_encrypted, invited_email_hash, status, invite_token, expires_at) VALUES (?, ?, ?, ?, 'pending', ?, ?)",
(team_id, role, None, email_hash, token, expires_at),
)
member_id = conn.execute("SELECT last_insert_rowid()").fetchone()[0]
# Encrypt with the actual member ID
enc_email = encrypt_entity("team_member", member_id, email)
conn.execute(
"UPDATE team_members SET invited_email_encrypted = ? WHERE id = ?",
(enc_email, member_id),
)
conn.commit()
# Send invite email
try:
from app.routes.email import NotificationService
domain = os.environ.get("FRONTEND_DOMAIN", "agentforms.io")
invite_url = f"https://{domain}/invite/{token}"
html_body = f"""
<h2>Join a team on AgentForms</h2>
<p>You've been invited to join <strong>{team["name"]}</strong> as a <strong>{role}</strong>.</p>
<p><a href="{invite_url}">Click here to accept the invitation</a></p>
<p>Or visit: {invite_url}</p>
"""
plain_body = f"Join a team on AgentForms\n\nYou've been invited to join {team['name']} as a {role}.\n\nAccept at: {invite_url}"
NotificationService.send_html(
to=email,
subject="You're invited to join a team on AgentForms",
html_body=html_body,
plain_body=plain_body,
)
except Exception as e:
print(f"[team] Failed to send invite email: {e}")
return {"token": token, "email": email}
finally:
if conn:
conn.close()
def accept_invite(token, user_id):
"""Accept a team invitation. Links user to team."""
conn = None
try:
conn = get_db()
member = conn.execute(
"SELECT * FROM team_members WHERE invite_token = ? AND status = 'pending'",
(token,),
).fetchone()
if not member:
return {"error": "Invalid or expired invite"}
# Check token expiration
expires_at = member.get("expires_at")
if expires_at:
expiry = datetime.fromisoformat(expires_at)
if datetime.now(UTC) > expiry:
return {"error": "This invite has expired"}
conn.execute(
"UPDATE team_members SET user_id = ?, status = 'active', joined_at = ?, invited_email = NULL, invited_email_encrypted = NULL, invited_email_hash = NULL, invite_token = NULL WHERE id = ?",
(user_id, datetime.now(UTC).isoformat(), member["id"]),
)
conn.commit()
return {"team_id": member["team_id"], "role": member["role"]}
finally:
if conn:
conn.close()
def remove_team_member(team_id, member_id, remover_user_id):
"""Remove a member from a team. Only owner/admin can do this."""
conn = None
try:
conn = get_db()
# Check remover has permission
remover = conn.execute(
"SELECT role FROM team_members WHERE team_id = ? AND user_id = ? AND status = 'active'",
(team_id, remover_user_id),
).fetchone()
if not remover or remover["role"] not in ("owner", "admin"):
return {"error": "Permission denied"}
# Prevent removing owner
target = conn.execute(
"SELECT role FROM team_members WHERE team_id = ? AND id = ?",
(team_id, member_id),
).fetchone()
if not target:
return {"error": "Member not found"}
if target["role"] == "owner":
return {"error": "Cannot remove team owner"}
conn.execute(
"DELETE FROM team_members WHERE team_id = ? AND id = ?",
(team_id, member_id),
)
conn.commit()
return {"success": True}
finally:
if conn:
conn.close()
def change_member_role(team_id, member_id, new_role, changer_user_id):
"""Change a member's role. Only owner/admin can do this."""
conn = None
try:
conn = get_db()
changer = conn.execute(
"SELECT role FROM team_members WHERE team_id = ? AND user_id = ? AND status = 'active'",
(team_id, changer_user_id),
).fetchone()
if not changer or changer["role"] not in ("owner", "admin"):
return {"error": "Permission denied"}
if new_role not in ("owner", "admin", "member", "viewer"):
return {"error": "Invalid role"}
conn.execute(
"UPDATE team_members SET role = ? WHERE team_id = ? AND id = ?",
(new_role, team_id, member_id),
)
conn.commit()
return {"success": True}
finally:
if conn:
conn.close()
def get_user_team_role(user_id, team_id):
"""Get a user's role in a team. Returns None if not a member."""
conn = None
try:
conn = get_db()
row = conn.execute(
"SELECT role FROM team_members WHERE team_id = ? AND user_id = ? AND status = 'active'",
(team_id, user_id),
).fetchone()
finally:
if conn:
conn.close()
return row["role"] if row else None
def delete_team(team_id, user_id):
"""Delete a team. Only owner can do this."""
conn = None
try:
conn = get_db()
member = conn.execute(
"SELECT role FROM team_members WHERE team_id = ? AND user_id = ? AND status = 'active'",
(team_id, user_id),
).fetchone()
if not member or member["role"] != "owner":
return {"error": "Only team owner can delete"}
# Unassign team sites
conn.execute("UPDATE sites SET team_id = NULL WHERE team_id = ?", (team_id,))
# Delete members
conn.execute("DELETE FROM team_members WHERE team_id = ?", (team_id,))
# Delete team
conn.execute("DELETE FROM teams WHERE id = ?", (team_id,))
conn.commit()
return {"success": True}
finally:
if conn:
conn.close()