import postgres from "postgres";
import type { Annotation, PolarisEvent, Project, Session, ParticipantId } from "../types";

export type Sql = postgres.Sql;

// --- Types ---

export interface Org {
  id: string;
  name: string;
  slug: string | null;
  domain: string | null;
  plan: string;
  slack_team_id: string | null;
  slack_bot_token: string | null;
  slack_system_channel_id: string | null;
  created_at: string;
}

export interface User {
  id: string;
  email: string;
  name: string;
  org_id: string;
  participant_id: string;
  created_at: string;
}

export interface ProjectMember {
  participant_id: string;
  role: string | null;
}

// --- Schema ---

export async function createDb(connectionString?: string): Promise<Sql> {
  const connStr = connectionString ?? process.env.DATABASE_URL ?? "postgres://polaris:polaris@localhost:5432/polaris";
  const sql = postgres(connStr);
  // Retry on transient startup errors (ECONNREFUSED, "starting up" 57P03).
  // Fail fast on auth/role/database errors — retrying won't help.
  const TRANSIENT_CODES = new Set(["57P03", "ECONNREFUSED"]);
  const FATAL_CODES = new Set(["28000", "28P01", "3D000"]);
  const MAX_ATTEMPTS = 10;
  const RETRY_DELAY_MS = 1000;
  for (let attempt = 1; attempt <= MAX_ATTEMPTS; attempt++) {
    try {
      await ensureSchema(sql);
      return sql;
    } catch (err: unknown) {
      const code = (err as { code?: string }).code ?? "";
      if (FATAL_CODES.has(code)) {
        throw new Error(
          `Database connection failed (${code}): ${(err as Error).message}\n` +
          `Hint: the role/password/database may not exist. ` +
          `Try: docker compose down -v && docker compose up -d --wait`
        );
      }
      if (!TRANSIENT_CODES.has(code) && attempt === MAX_ATTEMPTS) {
        throw err;
      }
      if (attempt < MAX_ATTEMPTS) {
        console.error(`[db] connection attempt ${attempt}/${MAX_ATTEMPTS} failed (${code}), retrying in ${RETRY_DELAY_MS}ms...`);
        await new Promise((r) => setTimeout(r, RETRY_DELAY_MS));
      } else {
        throw err;
      }
    }
  }
  // unreachable
  return sql;
}

// Arbitrary fixed key for the schema-init advisory lock (issue #117).
const SCHEMA_LOCK_KEY = 0x504c5253; // "PLRS"

export async function ensureSchema(sql: Sql): Promise<void> {
  // Serialize concurrent callers (api + web start together) with an advisory lock on a
  // reserved connection — concurrent CREATE TABLE IF NOT EXISTS races otherwise (issue #117).
  const lock = await sql.reserve();
  try {
    await lock`SELECT pg_advisory_lock(${SCHEMA_LOCK_KEY}::bigint)`;
    await createSchemaTables(sql);
  } finally {
    await lock`SELECT pg_advisory_unlock(${SCHEMA_LOCK_KEY}::bigint)`;
    lock.release();
  }
}

async function createSchemaTables(sql: Sql): Promise<void> {
  await sql`
    CREATE TABLE IF NOT EXISTS orgs (
      id TEXT PRIMARY KEY,
      name TEXT NOT NULL,
      slug TEXT UNIQUE,
      domain TEXT,
      plan TEXT NOT NULL DEFAULT 'free',
      slack_team_id TEXT,
      slack_bot_token TEXT,
      slack_system_channel_id TEXT,
      created_at TIMESTAMPTZ NOT NULL DEFAULT now()
    )
  `;

  // Migrate: add plan column if missing
  await sql`ALTER TABLE orgs ADD COLUMN IF NOT EXISTS plan TEXT NOT NULL DEFAULT 'free'`;

  await sql`
    CREATE TABLE IF NOT EXISTS users (
      id TEXT PRIMARY KEY,
      email TEXT NOT NULL UNIQUE,
      name TEXT NOT NULL,
      org_id TEXT NOT NULL REFERENCES orgs(id),
      participant_id TEXT NOT NULL,
      created_at TIMESTAMPTZ NOT NULL DEFAULT now()
    )
  `;

  await sql`
    CREATE TABLE IF NOT EXISTS schema_migrations (
      id TEXT PRIMARY KEY,
      applied_at TIMESTAMPTZ NOT NULL DEFAULT now()
    )
  `;

  await sql`
    CREATE TABLE IF NOT EXISTS projects (
      id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
      org_id TEXT NOT NULL REFERENCES orgs(id),
      name TEXT NOT NULL,
      slack_channel_id TEXT,
      slack_channel_name TEXT,
      created_at TIMESTAMPTZ NOT NULL DEFAULT now(),
      UNIQUE (org_id, name)
    )
  `;

  await sql`
    CREATE TABLE IF NOT EXISTS sessions (
      name TEXT NOT NULL,
      project_id UUID NOT NULL REFERENCES projects(id),
      org_id TEXT NOT NULL,
      driver TEXT,
      created_at TIMESTAMPTZ NOT NULL DEFAULT now(),
      PRIMARY KEY (project_id, name)
    )
  `;

  await sql`
    CREATE TABLE IF NOT EXISTS events (
      id UUID PRIMARY KEY,
      org_id TEXT NOT NULL,
      project_id UUID NOT NULL REFERENCES projects(id),
      session TEXT NOT NULL,
      timestamp TIMESTAMPTZ NOT NULL,
      source TEXT NOT NULL,
      sender TEXT NOT NULL,
      payload JSONB NOT NULL
    )
  `;

  await sql`
    CREATE INDEX IF NOT EXISTS idx_events_project ON events(project_id, timestamp)
  `;

  await sql`
    CREATE INDEX IF NOT EXISTS idx_events_session ON events(project_id, session, timestamp)
  `;

  // Curation annotations (stars, tags, decisions). Kept free of foreign keys.
  await sql`
    CREATE TABLE IF NOT EXISTS annotations (
      id UUID PRIMARY KEY,
      org_id TEXT NOT NULL,
      event_id UUID,
      project TEXT NOT NULL,
      session TEXT NOT NULL,
      participant_id TEXT,
      kind TEXT NOT NULL,
      value TEXT,
      created_at TIMESTAMPTZ NOT NULL DEFAULT now()
    )
  `;

  await sql`
    CREATE INDEX IF NOT EXISTS idx_annotations_session ON annotations(org_id, project, session)
  `;

  await sql`
    CREATE INDEX IF NOT EXISTS idx_annotations_event ON annotations(event_id)
  `;

  // Per-project ACL membership. Kept free of foreign keys (see annotations note above).
  await sql`
    CREATE TABLE IF NOT EXISTS project_members (
      project_id UUID,
      participant_id TEXT,
      role TEXT,
      PRIMARY KEY (project_id, participant_id)
    )
  `;

  await sql`
    CREATE TABLE IF NOT EXISTS plan_changes (
      id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
      org_id TEXT NOT NULL REFERENCES orgs(id),
      user_id TEXT NOT NULL REFERENCES users(id),
      from_plan TEXT NOT NULL,
      to_plan TEXT NOT NULL,
      created_at TIMESTAMPTZ NOT NULL DEFAULT now()
    )
  `;

  await runMigrations(sql);
}

// Additive, idempotent migrations. Each statement uses IF NOT EXISTS so it is safe to
// re-run on every startup; schema_migrations records what has been applied. NEVER drop a table.
async function runMigrations(sql: Sql): Promise<void> {
  const migrations: Array<{ id: string; run: () => Promise<unknown> }> = [
    {
      id: "001-sessions-label",
      run: () => sql`ALTER TABLE sessions ADD COLUMN IF NOT EXISTS label TEXT`,
    },
    {
      id: "002-projects-visibility",
      run: () => sql`ALTER TABLE projects ADD COLUMN IF NOT EXISTS visibility TEXT NOT NULL DEFAULT 'org'`,
    },
  ];
  for (const migration of migrations) {
    await migration.run();
    await sql`INSERT INTO schema_migrations (id) VALUES (${migration.id}) ON CONFLICT (id) DO NOTHING`;
  }
}

// --- Orgs ---

export async function createOrg(sql: Sql, id: string, name: string, domain?: string, plan?: string): Promise<Org> {
  const [row] = await sql`
    INSERT INTO orgs (id, name, domain, plan) VALUES (${id}, ${name}, ${domain ?? null}, ${plan ?? "free"})
    RETURNING *
  `;
  return { ...row, created_at: row.created_at.toISOString() } as Org;
}

export async function setOrgPlan(sql: Sql, orgId: string, fromPlan: string, toPlan: string, userId: string): Promise<void> {
  await sql`UPDATE orgs SET plan = ${toPlan} WHERE id = ${orgId}`;
  await sql`INSERT INTO plan_changes (org_id, user_id, from_plan, to_plan) VALUES (${orgId}, ${userId}, ${fromPlan}, ${toPlan})`;
}

export async function getOrg(sql: Sql, id: string): Promise<Org | null> {
  const [row] = await sql`SELECT * FROM orgs WHERE id = ${id}`;
  if (!row) return null;
  return { ...row, created_at: row.created_at.toISOString() } as Org;
}

export async function getOrgByDomain(sql: Sql, domain: string): Promise<Org | null> {
  const [row] = await sql`SELECT * FROM orgs WHERE domain = ${domain}`;
  if (!row) return null;
  return { ...row, created_at: row.created_at.toISOString() } as Org;
}

export async function setOrgSlack(sql: Sql, orgId: string, teamId: string, botToken: string, systemChannelId?: string, slug?: string): Promise<void> {
  if (slug) {
    await sql`UPDATE orgs SET slack_team_id = ${teamId}, slack_bot_token = ${botToken}, slack_system_channel_id = ${systemChannelId ?? null}, slug = ${slug} WHERE id = ${orgId}`;
  } else {
    await sql`UPDATE orgs SET slack_team_id = ${teamId}, slack_bot_token = ${botToken}, slack_system_channel_id = ${systemChannelId ?? null} WHERE id = ${orgId}`;
  }
}

export async function getOrgBySlug(sql: Sql, slug: string): Promise<Org | null> {
  const [row] = await sql`SELECT * FROM orgs WHERE slug = ${slug}`;
  if (!row) return null;
  return { ...row, created_at: row.created_at.toISOString() } as Org;
}

// --- Users ---

export async function createUser(sql: Sql, id: string, email: string, name: string, orgId: string, participantId: string): Promise<User> {
  const [row] = await sql`
    INSERT INTO users (id, email, name, org_id, participant_id) VALUES (${id}, ${email}, ${name}, ${orgId}, ${participantId})
    RETURNING *
  `;
  return { ...row, created_at: row.created_at.toISOString() } as User;
}

export async function getUser(sql: Sql, id: string): Promise<User | null> {
  const [row] = await sql`SELECT * FROM users WHERE id = ${id}`;
  if (!row) return null;
  return { ...row, created_at: row.created_at.toISOString() } as User;
}

export async function getUserByEmail(sql: Sql, email: string): Promise<User | null> {
  const [row] = await sql`SELECT * FROM users WHERE email = ${email}`;
  if (!row) return null;
  return { ...row, created_at: row.created_at.toISOString() } as User;
}

export async function upsertUser(sql: Sql, id: string, email: string, name: string, orgId: string, participantId: string): Promise<User> {
  const [row] = await sql`
    INSERT INTO users (id, email, name, org_id, participant_id) VALUES (${id}, ${email}, ${name}, ${orgId}, ${participantId})
    ON CONFLICT (email) DO UPDATE SET name = ${name}, org_id = ${orgId}, participant_id = ${participantId}
    RETURNING *
  `;
  return { ...row, created_at: row.created_at.toISOString() } as User;
}

export async function listUsers(sql: Sql, orgId: string): Promise<User[]> {
  const rows = await sql`SELECT * FROM users WHERE org_id = ${orgId} ORDER BY name ASC`;
  return rows.map((r) => ({ ...r, created_at: r.created_at.toISOString() }) as User);
}

export async function getRecentSignups(sql: Sql, since: Date, limit = 10): Promise<Array<User & { org_name: string }>> {
  const rows = await sql`
    SELECT u.*, o.name as org_name
    FROM users u JOIN orgs o ON u.org_id = o.id
    WHERE u.created_at >= ${since.toISOString()}
    ORDER BY u.created_at DESC
    LIMIT ${limit}
  `;
  return rows.map((r) => ({ ...r, created_at: r.created_at.toISOString(), org_name: r.org_name }) as User & { org_name: string });
}

// --- Projects (org-scoped) ---

// SELECT *  (not an explicit column list) so reads tolerate pre-visibility-migration rows.
function rowToProject(row: postgres.Row): Project {
  return {
    id: row.id,
    name: row.name,
    slack_channel_id: row.slack_channel_id ?? null,
    slack_channel_name: row.slack_channel_name ?? null,
    visibility: row.visibility ?? "org",
    created_at: row.created_at.toISOString(),
  };
}

export async function createProject(sql: Sql, orgId: string, name: string): Promise<Project> {
  const [row] = await sql`
    INSERT INTO projects (org_id, name) VALUES (${orgId}, ${name})
    RETURNING *
  `;
  return rowToProject(row);
}

export async function renameSession(sql: Sql, orgId: string, project: string, oldName: string, newName: string): Promise<void> {
  const result = await sql`
    UPDATE sessions SET name = ${newName}
    WHERE org_id = ${orgId} AND name = ${oldName}
      AND project_id = (SELECT id FROM projects WHERE org_id = ${orgId} AND name = ${project})
  `;
  if (result.count === 0) throw new Error(`Session not found: ${project}/${oldName}`);
}

export async function renameProject(sql: Sql, orgId: string, oldName: string, newName: string): Promise<void> {
  await sql`
    UPDATE projects SET name = ${newName}, slack_channel_name = ${newName}
    WHERE org_id = ${orgId} AND name = ${oldName}
  `;
}

export async function setProjectSlackChannel(sql: Sql, orgId: string, projectName: string, channelId: string, channelName?: string): Promise<void> {
  await sql`
    UPDATE projects SET slack_channel_id = ${channelId}, slack_channel_name = ${channelName ?? null}
    WHERE org_id = ${orgId} AND name = ${projectName}
  `;
}

export async function listProjects(sql: Sql, orgId: string): Promise<Project[]> {
  const rows = await sql`
    SELECT * FROM projects WHERE org_id = ${orgId} ORDER BY created_at ASC
  `;
  return rows.map(rowToProject);
}

export async function getProject(sql: Sql, orgId: string, name: string): Promise<Project | null> {
  const [row] = await sql`
    SELECT * FROM projects WHERE org_id = ${orgId} AND name = ${name}
  `;
  if (!row) return null;
  return rowToProject(row);
}

// --- Project visibility & members (ACL) ---

export async function setProjectVisibility(sql: Sql, orgId: string, project: string, visibility: string): Promise<void> {
  // Self-heal: resetTestData recreates projects without the visibility column, so make
  // sure it exists (idempotent) before writing.
  await sql`ALTER TABLE projects ADD COLUMN IF NOT EXISTS visibility TEXT NOT NULL DEFAULT 'org'`;
  await sql`
    UPDATE projects SET visibility = ${visibility}
    WHERE org_id = ${orgId} AND name = ${project}
  `;
}

export async function addProjectMember(sql: Sql, orgId: string, project: string, participantId: string, role?: string): Promise<void> {
  await sql`
    INSERT INTO project_members (project_id, participant_id, role)
    SELECT p.id, ${participantId}, ${role ?? null}
    FROM projects p WHERE p.org_id = ${orgId} AND p.name = ${project}
    ON CONFLICT (project_id, participant_id) DO UPDATE SET role = EXCLUDED.role
  `;
}

export async function removeProjectMember(sql: Sql, orgId: string, project: string, participantId: string): Promise<void> {
  await sql`
    DELETE FROM project_members
    WHERE participant_id = ${participantId}
      AND project_id = (SELECT id FROM projects WHERE org_id = ${orgId} AND name = ${project})
  `;
}

export async function listProjectMembers(sql: Sql, orgId: string, project: string): Promise<ProjectMember[]> {
  const rows = await sql`
    SELECT m.participant_id, m.role
    FROM project_members m
    JOIN projects p ON m.project_id = p.id
    WHERE p.org_id = ${orgId} AND p.name = ${project}
    ORDER BY m.participant_id ASC
  `;
  return rows.map((r) => ({ participant_id: r.participant_id, role: r.role ?? null }));
}

export async function userCanAccessProject(sql: Sql, orgId: string, project: string, participantId: string | null): Promise<boolean> {
  // Default-open: anonymous callers, unknown projects, 'org' visibility, and any schema
  // created before the visibility migration / project_members table all mean accessible.
  if (!participantId) return true;
  try {
    const [row] = await sql`SELECT * FROM projects WHERE org_id = ${orgId} AND name = ${project}`;
    if (!row) return true;
    if ((row.visibility ?? "org") !== "members") return true;
    const members = await sql`
      SELECT 1 FROM project_members WHERE project_id = ${row.id} AND participant_id = ${participantId}
    `;
    return members.length > 0;
  } catch {
    return true;
  }
}

// --- Sessions (org-scoped) ---

export async function createSession(
  sql: Sql,
  orgId: string,
  project: string,
  name: string,
  driver: ParticipantId | null
): Promise<Session> {
  // RETURNING * so this tolerates pre-label-migration session tables (label -> null).
  const [row] = await sql`
    INSERT INTO sessions (name, project_id, org_id, driver)
    SELECT ${name}, p.id, ${orgId}, ${driver}
    FROM projects p WHERE p.org_id = ${orgId} AND p.name = ${project}
    RETURNING *
  `;
  if (!row) throw new Error(`Project not found: ${project}`);
  return {
    name: row.name,
    project,
    driver: row.driver,
    label: row.label ?? null,
    created_at: row.created_at.toISOString(),
  };
}

export async function getSession(sql: Sql, orgId: string, project: string, name: string): Promise<Session | null> {
  // s.* so this tolerates pre-label-migration session tables (label -> null).
  const [row] = await sql`
    SELECT s.*, p.name as project
    FROM sessions s
    JOIN projects p ON s.project_id = p.id
    WHERE s.org_id = ${orgId} AND p.name = ${project} AND s.name = ${name}
  `;
  if (!row) return null;
  return {
    name: row.name,
    project: row.project,
    driver: row.driver,
    label: row.label ?? null,
    created_at: row.created_at.toISOString(),
  };
}

export async function listSessions(sql: Sql, orgId: string, project?: string): Promise<Session[]> {
  const rows = project
    ? await sql`
        SELECT s.*, p.name as project
        FROM sessions s
        JOIN projects p ON s.project_id = p.id
        WHERE s.org_id = ${orgId} AND p.name = ${project}
        ORDER BY s.created_at ASC
      `
    : await sql`
        SELECT s.*, p.name as project
        FROM sessions s
        JOIN projects p ON s.project_id = p.id
        WHERE s.org_id = ${orgId}
        ORDER BY s.created_at ASC
      `;
  return rows.map((row) => ({
    name: row.name,
    project: row.project,
    driver: row.driver,
    label: row.label ?? null,
    created_at: row.created_at.toISOString(),
  }));
}

export async function getSessionPromptCounts(sql: Sql, orgId: string): Promise<Map<string, number>> {
  const rows = await sql`
    SELECT p.name as project, e.session, count(*)::int as count
    FROM events e
    JOIN projects p ON e.project_id = p.id
    WHERE e.org_id = ${orgId} AND e.payload->>'hook_event_name' = 'UserPromptSubmit'
    GROUP BY p.name, e.session
  `;
  const counts = new Map<string, number>();
  for (const row of rows) {
    counts.set(`${row.project}/${row.session}`, row.count);
  }
  return counts;
}

export async function getDailyPromptCounts(sql: Sql, orgId: string, days = 14): Promise<Array<{ date: string; sender: string; count: number }>> {
  const rows = await sql`
    SELECT date_trunc('day', e.timestamp)::date as date, e.sender, count(*)::int as count
    FROM events e
    WHERE e.org_id = ${orgId} AND e.payload->>'hook_event_name' = 'UserPromptSubmit'
      AND e.timestamp >= now() - ${days + ' days'}::interval
    GROUP BY date, e.sender
    ORDER BY date ASC
  `;
  return rows.map((r) => ({ date: r.date.toISOString().slice(0, 10), sender: r.sender, count: r.count }));
}

export async function setDriver(sql: Sql, orgId: string, project: string, session: string, driver: ParticipantId): Promise<void> {
  await sql`
    UPDATE sessions SET driver = ${driver}
    WHERE org_id = ${orgId} AND name = ${session}
      AND project_id = (SELECT id FROM projects WHERE org_id = ${orgId} AND name = ${project})
  `;
}

export async function clearDriver(sql: Sql, orgId: string, project: string, session: string): Promise<void> {
  await sql`
    UPDATE sessions SET driver = NULL
    WHERE org_id = ${orgId} AND name = ${session}
      AND project_id = (SELECT id FROM projects WHERE org_id = ${orgId} AND name = ${project})
  `;
}

export async function setSessionLabel(sql: Sql, orgId: string, project: string, session: string, label: string): Promise<void> {
  await sql`
    UPDATE sessions SET label = ${label}
    WHERE org_id = ${orgId} AND name = ${session}
      AND project_id = (SELECT id FROM projects WHERE org_id = ${orgId} AND name = ${project})
  `;
}

// --- Events (org-scoped) ---

export async function pushEvent(sql: Sql, orgId: string, event: PolarisEvent): Promise<void> {
  const result = await sql`
    INSERT INTO events (id, org_id, project_id, session, timestamp, source, sender, payload)
    SELECT ${event.id}, ${orgId}, p.id, ${event.session}, ${event.timestamp}, ${event.source}, ${event.sender}, ${sql.json(event.payload)}
    FROM projects p WHERE p.org_id = ${orgId} AND p.name = ${event.project}
  `;
  // Realtime backbone: notify listeners (server WS/SSE fan-out, Slack bridge) with the
  // event id only — tiny payload; listeners fetch the full event via getEventById.
  if (result.count > 0) await sql.notify("polaris_event", event.id);
}

function rowToEvent(row: postgres.Row): PolarisEvent {
  return {
    id: row.id,
    project: row.project,
    session: row.session,
    timestamp: row.timestamp.toISOString(),
    source: row.source as PolarisEvent["source"],
    sender: row.sender as ParticipantId,
    payload: row.payload as PolarisEvent["payload"],
  };
}

// Lookup for LISTEN('polaris_event') subscribers. Returns org_id too so multi-org
// consumers (e.g. the Slack bridge) can route without a second query.
export async function getEventById(sql: Sql, id: string): Promise<(PolarisEvent & { org_id: string }) | null> {
  // NOTIFY payloads are untyped strings; a non-UUID would fail the uuid cast, so treat it as not-found.
  if (!/^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$/i.test(id)) return null;
  const [row] = await sql`
    SELECT e.id, e.org_id, p.name as project, e.session, e.timestamp, e.source, e.sender, e.payload
    FROM events e
    JOIN projects p ON e.project_id = p.id
    WHERE e.id = ${id}
  `;
  if (!row) return null;
  return { ...rowToEvent(row), org_id: row.org_id };
}

export async function getProjectEvents(sql: Sql, orgId: string, project: string): Promise<PolarisEvent[]> {
  const rows = await sql`
    SELECT e.id, p.name as project, e.session, e.timestamp, e.source, e.sender, e.payload
    FROM events e
    JOIN projects p ON e.project_id = p.id
    WHERE e.org_id = ${orgId} AND p.name = ${project}
    ORDER BY e.timestamp ASC
  `;
  return rows.map(rowToEvent);
}

export async function getSessionEvents(sql: Sql, orgId: string, project: string, session: string): Promise<PolarisEvent[]> {
  const rows = await sql`
    SELECT e.id, p.name as project, e.session, e.timestamp, e.source, e.sender, e.payload
    FROM events e
    JOIN projects p ON e.project_id = p.id
    WHERE e.org_id = ${orgId} AND p.name = ${project} AND e.session = ${session}
    ORDER BY e.timestamp ASC
  `;
  return rows.map(rowToEvent);
}

export async function getSessionEventsPage(
  sql: Sql,
  orgId: string,
  project: string,
  session: string,
  opts?: { limit?: number; before?: string }
): Promise<{ events: PolarisEvent[]; nextCursor: string | null }> {
  const limit = Math.min(Math.max(1, Math.floor(opts?.limit ?? 100)), 500);
  // Cursor is `${iso_timestamp}|${id}`; keyset pagination on (timestamp, id) DESC.
  let beforeTs: string | null = null;
  let beforeId: string | null = null;
  if (opts?.before) {
    const sep = opts.before.lastIndexOf("|");
    if (sep > 0) {
      beforeTs = opts.before.slice(0, sep);
      beforeId = opts.before.slice(sep + 1);
    }
  }
  const rows = beforeTs && beforeId
    ? await sql`
        SELECT e.id, p.name as project, e.session, e.timestamp, e.source, e.sender, e.payload
        FROM events e
        JOIN projects p ON e.project_id = p.id
        WHERE e.org_id = ${orgId} AND p.name = ${project} AND e.session = ${session}
          AND (e.timestamp, e.id) < (${beforeTs}::timestamptz, ${beforeId}::uuid)
        ORDER BY e.timestamp DESC, e.id DESC
        LIMIT ${limit}
      `
    : await sql`
        SELECT e.id, p.name as project, e.session, e.timestamp, e.source, e.sender, e.payload
        FROM events e
        JOIN projects p ON e.project_id = p.id
        WHERE e.org_id = ${orgId} AND p.name = ${project} AND e.session = ${session}
        ORDER BY e.timestamp DESC, e.id DESC
        LIMIT ${limit}
      `;
  const events = rows.map(rowToEvent);
  const oldest = events[events.length - 1];
  const nextCursor = events.length === limit && oldest ? `${oldest.timestamp}|${oldest.id}` : null;
  return { events, nextCursor };
}

// Query-time full-text search (explicit 'english'::regconfig form, immutable-safe; no index yet).
// FOLLOW-UP: trigger-maintained indexed search_tsv before scale.
export async function searchEvents(
  sql: Sql,
  orgId: string,
  opts: { q: string; project?: string; session?: string; sender?: string; source?: string; tag?: string; limit?: number; participantId?: string | null }
): Promise<{ results: Array<{ event: PolarisEvent; snippet: string }> }> {
  const limit = Math.min(Math.max(1, Math.floor(opts.limit ?? 50)), 100);
  // Restrict to projects the caller may access — without this an unfiltered search
  // would leak 'members'-restricted projects. (no participant -> default-open)
  const aclFilter = opts.participantId
    ? sql`AND (COALESCE(p.visibility, 'org') <> 'members' OR EXISTS (
        SELECT 1 FROM project_members pm WHERE pm.project_id = p.id AND pm.participant_id = ${opts.participantId}
      ))`
    : sql``;
  const rows = await sql`
    SELECT e.id, p.name as project, e.session, e.timestamp, e.source, e.sender, e.payload,
      ts_headline('english'::regconfig, t.text, q.query) as snippet,
      ts_rank(to_tsvector('english'::regconfig, t.text), q.query) as score
    FROM events e
    JOIN projects p ON e.project_id = p.id
    CROSS JOIN LATERAL (
      SELECT coalesce(e.payload->>'prompt', '') || ' ' || coalesce(e.payload->>'stop_response', '') || ' ' ||
        coalesce(e.payload->>'last_assistant_message', '') || ' ' || coalesce(e.payload->>'content', '') as text
    ) t
    CROSS JOIN LATERAL (SELECT websearch_to_tsquery('english', ${opts.q}) as query) q
    WHERE e.org_id = ${orgId}
      AND to_tsvector('english'::regconfig, t.text) @@ q.query
      ${opts.project ? sql`AND p.name = ${opts.project}` : sql``}
      ${opts.session ? sql`AND e.session = ${opts.session}` : sql``}
      ${opts.sender ? sql`AND e.sender = ${opts.sender}` : sql``}
      ${opts.source ? sql`AND e.source = ${opts.source}` : sql``}
      ${opts.tag ? sql`AND EXISTS (
        SELECT 1 FROM annotations a
        WHERE a.event_id = e.id AND a.org_id = ${orgId} AND a.kind = 'tag' AND a.value = ${opts.tag}
      )` : sql``}
      ${aclFilter}
    ORDER BY score DESC, e.timestamp DESC
    LIMIT ${limit}
  `;
  return {
    results: rows.map((row) => ({ event: rowToEvent(row), snippet: row.snippet as string })),
  };
}

export async function getOrgEventsSince(sql: Sql, orgId: string, since: string): Promise<PolarisEvent[]> {
  const rows = await sql`
    SELECT e.id, p.name as project, e.session, e.timestamp, e.source, e.sender, e.payload
    FROM events e
    JOIN projects p ON e.project_id = p.id
    WHERE e.org_id = ${orgId} AND e.timestamp > ${since}
    ORDER BY e.timestamp ASC
  `;
  return rows.map(rowToEvent);
}

export async function getEventsSince(sql: Sql, orgId: string, project: string, since: string): Promise<PolarisEvent[]> {
  const rows = await sql`
    SELECT e.id, p.name as project, e.session, e.timestamp, e.source, e.sender, e.payload
    FROM events e
    JOIN projects p ON e.project_id = p.id
    WHERE e.org_id = ${orgId} AND p.name = ${project} AND e.timestamp > ${since}
    ORDER BY e.timestamp ASC
  `;
  return rows.map(rowToEvent);
}

// --- Annotations (org-scoped curation: stars, tags, decisions) ---

function rowToAnnotation(row: postgres.Row): Annotation {
  return {
    id: row.id,
    event_id: row.event_id ?? null,
    project: row.project,
    session: row.session,
    participant_id: row.participant_id ?? null,
    kind: row.kind as Annotation["kind"],
    value: row.value ?? null,
    created_at: row.created_at.toISOString(),
  };
}

export async function addAnnotation(
  sql: Sql,
  orgId: string,
  annotation: {
    event_id?: string | null;
    project: string;
    session: string;
    participant_id?: string | null;
    kind: Annotation["kind"];
    value?: string | null;
  }
): Promise<{ id: string }> {
  const id = crypto.randomUUID();
  await sql`
    INSERT INTO annotations (id, org_id, event_id, project, session, participant_id, kind, value)
    VALUES (${id}, ${orgId}, ${annotation.event_id ?? null}, ${annotation.project}, ${annotation.session},
      ${annotation.participant_id ?? null}, ${annotation.kind}, ${annotation.value ?? null})
  `;
  return { id };
}

export async function listSessionAnnotations(sql: Sql, orgId: string, project: string, session: string): Promise<Annotation[]> {
  const rows = await sql`
    SELECT * FROM annotations
    WHERE org_id = ${orgId} AND project = ${project} AND session = ${session}
    ORDER BY created_at ASC
  `;
  return rows.map(rowToAnnotation);
}

export async function listDecisions(sql: Sql, orgId: string, opts?: { project?: string; limit?: number; participantId?: string | null }): Promise<Annotation[]> {
  const limit = Math.min(Math.max(1, Math.floor(opts?.limit ?? 100)), 500);
  // Exclude decisions from 'members'-restricted projects the caller isn't in
  // (default-open: missing project row / 'org' visibility / member -> kept).
  const aclFilter = opts?.participantId
    ? sql`AND NOT EXISTS (
        SELECT 1 FROM projects p
        WHERE p.org_id = annotations.org_id AND p.name = annotations.project
          AND COALESCE(p.visibility, 'org') = 'members'
          AND NOT EXISTS (
            SELECT 1 FROM project_members pm WHERE pm.project_id = p.id AND pm.participant_id = ${opts.participantId}
          )
      )`
    : sql``;
  const rows = await sql`
    SELECT * FROM annotations
    WHERE org_id = ${orgId} AND kind = 'decision'
      ${opts?.project ? sql`AND project = ${opts.project}` : sql``}
      ${aclFilter}
    ORDER BY created_at DESC
    LIMIT ${limit}
  `;
  return rows.map(rowToAnnotation);
}

export async function deleteAnnotation(sql: Sql, orgId: string, id: string): Promise<void> {
  await sql`DELETE FROM annotations WHERE org_id = ${orgId} AND id = ${id}`;
}