import { DurableObject } from "cloudflare:workers";
import { userInbox } from "../user-inbox";
import { activityTouches, indexActivity } from "../activity-index";
import { configurePerf } from "../perf";
import {
  armsPreciseRepoTask,
  armsRepoHousekeeping,
  markRepo,
  nextRepoDue,
  repoMarksOf,
  runRepoHousekeeping,
  type LastRuns,
} from "./housekeeping";
import type { InvolvementEntry, InvolvementRole } from "../involvement";
import { ScopedStore } from "../scoped-db/store";
import { REPO_SCHEMA } from "../scoped-db/schema.gen";
import { REPO_SCHEMA_EXTRA } from "../scoped-db/schema-extra";
import type {
  ScopedQueryResult,
  ScopedRunResult,
  ScopedSqlHost,
  ScopedStatement,
  ScopedValue,
} from "../scoped-db/types";

const SYNC_CHUNK = 50;
const SCOPE_KEY = "scope-id";
const INDEX_TIMEOUT_MS = 2000;
const RUNS_KEY = "housekeeping-runs";
const INVOLVEMENT_DDL = `CREATE TABLE IF NOT EXISTS involvement_published (
  issue_id TEXT PRIMARY KEY,
  users TEXT NOT NULL
)`;

type IssueRow = {
  id: string;
  number: number;
  is_pull: number;
  state: string;
  title: string;
  author_id: string;
  updated_at: number;
  hidden_at: number | null;
};

/** Who is involved in one issue and how, as of this sync. */
type Involvement = { entry: InvolvementEntry; users: Map<string, InvolvementRole[]> };

/**
 * One per repository: the home for everything that belongs to it (docs/state-layout.md). Its SQLite holds
 * the repository-scoped tables (schema.gen.ts); callers reach them through `repoDb(repoId)`, which speaks the
 * same first/all/run/batch API as D1.
 *
 * It also fans the involvement index out: it remembers which users each issue was published to, so a sync
 * updates their UserInbox and removes the issue from anyone no longer involved.
 */
export class RepoState extends DurableObject<Env> implements ScopedSqlHost {
  private readonly store = new ScopedStore(this.ctx.storage, [...REPO_SCHEMA, INVOLVEMENT_DDL, ...REPO_SCHEMA_EXTRA]);

  async sqlAll(sql: string, params: ScopedValue[]): Promise<Record<string, ScopedValue>[]> {
    await this.store.prepare();
    return this.store.all(sql, params);
  }

  async sqlQuery(sql: string, params: ScopedValue[]): Promise<ScopedQueryResult> {
    await this.store.prepare();
    return this.store.query(sql, params);
  }

  async sqlRun(sql: string, params: ScopedValue[], scopeId?: string): Promise<ScopedRunResult> {
    await this.store.prepare();
    const result = this.store.run(sql, params);
    await this.indexActivity([{ sql, params }]);
    if (armsRepoHousekeeping(sql)) {
      markRepo(this.store.sql, repoMarksOf(sql), Date.now());
      await this.arm(scopeId, armsPreciseRepoTask(sql));
    }
    return result;
  }

  async sqlBatch(statements: ScopedStatement[], scopeId?: string): Promise<ScopedRunResult[]> {
    await this.store.prepare();
    const results = this.store.batch(statements);
    await this.indexActivity(statements);
    if (statements.some((statement) => armsRepoHousekeeping(statement.sql))) {
      markRepo(this.store.sql, [...new Set(statements.flatMap((statement) => repoMarksOf(statement.sql)))], Date.now());
      await this.arm(
        scopeId,
        statements.some((statement) => armsPreciseRepoTask(statement.sql)),
      );
    }
    return results;
  }

  private scopeId: string | null = null;
  private lastRuns: LastRuns | null = null;

  private async ownScope(scopeId?: string): Promise<string | null> {
    if (scopeId && scopeId !== this.scopeId) {
      this.scopeId = scopeId;
      await this.ctx.storage.put(SCOPE_KEY, scopeId);
    }
    this.scopeId ??= (await this.ctx.storage.get<string>(SCOPE_KEY)) ?? null;
    return this.scopeId;
  }

  private async runs(): Promise<LastRuns> {
    this.lastRuns ??= (await this.ctx.storage.get<LastRuns>(RUNS_KEY)) ?? {};
    return this.lastRuns;
  }

  /**
   * Sets the alarm no later than the next due task; a later alarm is pulled in, an earlier one kept. Rows that are
   * only garbage later can't be due before a pending alarm (which recomputes when it fires), so they skip the
   * due queries unless no alarm is set.
   */
  private async arm(scopeId?: string, precise = true): Promise<void> {
    await this.ownScope(scopeId);
    const current = await this.ctx.storage.getAlarm();
    if (current !== null && !precise) return;
    const due = nextRepoDue(this.store.sql, Date.now(), await this.runs());
    if (due === null) return;
    if (current === null || current > due) await this.ctx.storage.setAlarm(due);
  }

  /** For the one-time backfill of repositories whose rows predate the alarm. */
  async armHousekeeping(scopeId: string): Promise<void> {
    await this.store.prepare();
    await this.arm(scopeId);
  }

  /** Runs the repository's due time-driven work (repo-state/housekeeping.ts), then re-arms. */
  override async alarm(): Promise<void> {
    configurePerf((this.env as { PERF_SAMPLE_RATE?: string }).PERF_SAMPLE_RATE);
    await this.store.prepare();
    const scopeId = await this.ownScope();
    if (!scopeId) return;
    const runs = await this.runs();
    await runRepoHousekeeping(scopeId, this.store.sql, Date.now(), runs);
    await this.ctx.storage.put(RUNS_KEY, runs);
    const due = nextRepoDue(this.store.sql, Date.now(), runs);
    if (due !== null) await this.ctx.storage.setAlarm(due);
  }

  /**
   * Feeds each actor's activity index (activity-index.ts) before the write is acknowledged: account deletion and
   * the contribution graph rely on it. Bounded by a timeout; touches that don't make it are queued in
   * `activity_index_pending`, which the repository's alarm retries, so the index is never silently short.
   */
  private async indexActivity(statements: ScopedStatement[]): Promise<void> {
    const touches = activityTouches(statements);
    if (touches.length === 0) return;
    const indexed = await Promise.race([
      indexActivity(touches).then(
        () => true,
        () => false,
      ),
      new Promise<boolean>((resolve) => setTimeout(() => resolve(false), INDEX_TIMEOUT_MS)),
    ]);
    if (indexed) return;
    this.store.batch(
      touches.map((touch) => ({
        sql: "INSERT INTO activity_index_pending (actor_id, repo_id, at, counted) VALUES (?, ?, ?, ?)",
        params: [touch.actorId, touch.repoId, touch.at, touch.counted ? 1 : 0],
      })),
    );
    await this.arm();
  }

  /** Republishes these issues to everyone involved in them; idempotent, so callers may repeat it freely. */
  async syncIssues(repoId: string, issueIds: string[]): Promise<void> {
    await this.store.prepare();
    for (let start = 0; start < issueIds.length; start += SYNC_CHUNK) {
      await this.publish(repoId, issueIds.slice(start, start + SYNC_CHUNK));
    }
  }

  /** The repository is gone: pull its issues out of every inbox they were published to, then forget everything. */
  async destroy(repoId: string): Promise<void> {
    await this.store.prepare();
    const users = new Set<string>();
    for (const row of this.store.all("SELECT users FROM involvement_published", [])) {
      for (const user of JSON.parse(String(row.users)) as string[]) users.add(user);
    }
    await Promise.all([...users].map((user) => userInbox(user).dropRepoInvolvement(repoId)));
    await this.store.wipe();
  }

  private async publish(repoId: string, issueIds: string[]): Promise<void> {
    if (issueIds.length === 0) return;
    const current = this.loadInvolvement(repoId, issueIds);
    const sql = this.store.sql;
    const writes: Array<Promise<void>> = [];
    for (const issueId of issueIds) {
      const previous = new Set<string>(
        JSON.parse(
          sql
            .exec<{ users: string }>("SELECT users FROM involvement_published WHERE issue_id = ?", issueId)
            .toArray()[0]?.users ?? "[]",
        ) as string[],
      );
      const now = current.get(issueId);
      if (now) {
        for (const [user, roles] of now.users) {
          writes.push(userInbox(user).putInvolvement(now.entry, roles));
          previous.delete(user);
        }
      }
      for (const user of previous) writes.push(userInbox(user).dropInvolvement([issueId]));
      const users = [...(now?.users.keys() ?? [])];
      if (users.length === 0) sql.exec("DELETE FROM involvement_published WHERE issue_id = ?", issueId);
      else {
        sql.exec(
          `INSERT INTO involvement_published (issue_id, users) VALUES (?, ?)
           ON CONFLICT (issue_id) DO UPDATE SET users = excluded.users`,
          issueId,
          JSON.stringify(users),
        );
      }
    }
    await Promise.all(writes);
  }

  /** Reads this repository's own tables; hidden (moderated) items publish to nobody. */
  private loadInvolvement(repoId: string, issueIds: string[]): Map<string, Involvement> {
    const ids = issueIds.map(() => "?").join(", ");
    const issues = this.store.all(
      `SELECT id, number, is_pull, state, title, author_id, updated_at, hidden_at FROM issues WHERE id IN (${ids})`,
      issueIds,
    ) as unknown as IssueRow[];
    const assignees = this.store.all(
      `SELECT issue_id, user_id FROM issue_assignees WHERE issue_id IN (${ids})`,
      issueIds,
    ) as Array<{ issue_id: string; user_id: string }>;
    const reviewers = this.store.all(
      `SELECT issue_id, user_id FROM review_requests WHERE issue_id IN (${ids})`,
      issueIds,
    ) as Array<{ issue_id: string; user_id: string }>;
    const out = new Map<string, Involvement>();
    for (const issue of issues) {
      if (issue.hidden_at !== null) continue;
      const users = new Map<string, InvolvementRole[]>();
      const add = (user: string, role: InvolvementRole) => users.set(user, [...(users.get(user) ?? []), role]);
      add(issue.author_id, "author");
      for (const row of assignees) if (row.issue_id === issue.id) add(row.user_id, "assignee");
      for (const row of reviewers) if (row.issue_id === issue.id) add(row.user_id, "reviewer");
      out.set(issue.id, {
        entry: {
          issueId: issue.id,
          repoId,
          number: issue.number,
          isPull: issue.is_pull === 1,
          state: issue.state,
          title: issue.title,
          updatedAt: issue.updated_at,
        },
        users,
      });
    }
    return out;
  }
}