import { DurableObject, env } from "cloudflare:workers";
import { batch, stmt } from "../db";
import { branchSha, listRefs } from "../git/refs";
import { withStore } from "../git/store";
import { loadRepoById } from "../repos";
import { CommitIndex } from "./commit-index";
import type { ContributorRow, EmailDay } from "./commit-index";
import { fastForward, toIndexed } from "./fast-forward";

const STATE_KEY = "state";
const LOG_PAGE = 1000;
const PAGES_PER_ALARM = 5;
const MAX_INDEXED_COMMITS = 300_000;
const PUBLISH_CHUNK = 400;
const ROWS_PER_INSERT = 20;
const RETRY_ON_ERROR_MS = 60_000;
const WEEKS = 52;
const WEEK_MS = 7 * 24 * 60 * 60 * 1000;
const MAX_CONTRIBUTORS = 100;
const CONTRIBUTOR_SPARKLINES = 25;
const MAX_DAY_COMMITS = 20;

type Rebuild = { target: string; gen: number; offset: number };

type State = {
  repoId: string;
  branch: string | null;
  head: string | null;
  gen: number;
  rebuild: Rebuild | null;
  truncated: boolean;
  /** Bumped when indexed data gains fields (e.g. co-authors); an older index is rebuilt once. */
  schema?: number;
};

const INDEX_SCHEMA = 2;

export type InsightsSummary = {
  status: "empty" | "indexing" | "ready";
  head: string | null;
  totalCommits: number;
  truncated: boolean;
  weekly: number[];
  weekStart: number;
  contributors: Array<ContributorRow & { weekly: number[] }>;
};

async function publishRows(repoId: string, rows: EmailDay[]): Promise<void> {
  const deletes = rows.filter((row) => row.commits === 0);
  const upserts = rows.filter((row) => row.commits > 0);
  const statements = deletes.map((row) =>
    stmt("DELETE FROM commit_email_days WHERE repo_id = ? AND email = ? AND day = ?", repoId, row.email, row.day),
  );
  for (let start = 0; start < upserts.length; start += ROWS_PER_INSERT) {
    const group = upserts.slice(start, start + ROWS_PER_INSERT);
    statements.push(
      stmt(
        `INSERT INTO commit_email_days (repo_id, email, day, commits) VALUES ${group.map(() => "(?, ?, ?, ?)").join(", ")}
         ON CONFLICT (repo_id, email, day) DO UPDATE SET commits = excluded.commits`,
        ...group.flatMap((row) => [repoId, row.email, row.day, row.commits]),
      ),
    );
  }
  if (statements.length > 0) await batch(statements);
}

/**
 * One instance per repository: keeps the default branch's commit index for Insights and publishes per-email
 * daily counts to D1 for profile contribution graphs. Kicks are idempotent; the alarm reconciles the index
 * with the current default-branch head, so missed or duplicate events only cost a no-op pass.
 */
export class RepoInsights extends DurableObject<Env> {
  private indexState: CommitIndex | null = null;

  private get index(): CommitIndex {
    this.indexState ??= new CommitIndex(this.ctx.storage.sql);
    return this.indexState;
  }

  private async state(repoId: string): Promise<State> {
    const stored = await this.ctx.storage.get<State>(STATE_KEY);
    return stored ?? { repoId, branch: null, head: null, gen: 0, rebuild: null, truncated: false };
  }

  async kick(repoId: string): Promise<void> {
    const state = await this.state(repoId);
    await this.ctx.storage.put(STATE_KEY, { ...state, repoId });
    await this.ctx.storage.setAlarm(Date.now());
  }

  async summary(repoId: string): Promise<InsightsSummary> {
    const state = await this.state(repoId);
    const neverReconciled = state.branch === null;
    if (neverReconciled) await this.kick(repoId);
    const now = Date.now();
    const contributors = this.index.contributors(now - WEEKS * WEEK_MS, MAX_CONTRIBUTORS);
    const status = state.rebuild || neverReconciled ? "indexing" : state.head ? "ready" : "empty";
    return {
      status,
      head: state.head,
      totalCommits: this.index.count(),
      truncated: state.truncated,
      weekly: this.index.weekly(now, WEEKS),
      weekStart: now - WEEKS * WEEK_MS,
      contributors: contributors.map((row, position) => ({
        ...row,
        weekly: position < CONTRIBUTOR_SPARKLINES ? this.index.contributorWeekly(row.email, now, WEEKS) : [],
      })),
    };
  }

  async commitsOn(emails: string[], day: string, limit: number): Promise<Array<{ sha: string; authored_at: number }>> {
    return this.index.commitsOn(emails, day, Math.min(Math.max(1, limit), MAX_DAY_COMMITS));
  }

  override async alarm(): Promise<void> {
    const stored = await this.ctx.storage.get<State>(STATE_KEY);
    if (!stored) return;
    try {
      const again = await this.reconcile(stored);
      if (again) await this.ctx.storage.setAlarm(Date.now());
    } catch (error) {
      console.error("Repo insights pass failed", stored.repoId, error);
      await this.ctx.storage.setAlarm(Date.now() + RETRY_ON_ERROR_MS);
    }
  }

  /** Returns whether another pass is needed (rebuild pages or publish chunks left). */
  private async reconcile(initial: State): Promise<boolean> {
    let state = initial;
    const save = async (next: State) => {
      state = next;
      await this.ctx.storage.put(STATE_KEY, next);
    };
    const repo = await loadRepoById(state.repoId);
    if (!repo) {
      await this.ctx.storage.deleteAll();
      this.indexState = null;
      return false;
    }
    const head = repo.status === "ready" ? branchSha(await listRefs(repo), repo.default_branch) : null;
    if (head === null) {
      this.index.clear();
      await save({ ...state, branch: repo.default_branch, head: null, rebuild: null, truncated: false });
      return this.publish(state.repoId);
    }
    // An abandoned rebuild leaves rows that are not closed under ancestry, so only another rebuild may follow it.
    if (state.rebuild && state.rebuild.target !== head) {
      await save({
        ...state,
        branch: repo.default_branch,
        rebuild: { target: head, gen: state.rebuild.gen + 1, offset: 0 },
      });
    }
    if (!state.rebuild) {
      const current = state.schema === INDEX_SCHEMA;
      if (current && state.head === head && state.branch === repo.default_branch) return this.publish(state.repoId);
      const oldHead = state.head;
      const fresh =
        current && oldHead && state.branch === repo.default_branch
          ? await withStore(repo.artifacts_name, (store) =>
              fastForward(store, (sha) => this.index.has(sha), oldHead, head),
            )
          : null;
      if (fresh) {
        this.ctx.storage.transactionSync(() => this.index.upsert(fresh, state.gen));
        await save({ ...state, head });
        return this.publish(state.repoId);
      }
      await save({ ...state, branch: repo.default_branch, rebuild: { target: head, gen: state.gen + 1, offset: 0 } });
    }
    const rebuild = state.rebuild;
    if (!rebuild) return false;
    let offset = rebuild.offset;
    let done = false;
    await withStore(repo.artifacts_name, async (store) => {
      for (let page = 0; page < PAGES_PER_ALARM && !done; page++) {
        const commits = await store.log(rebuild.target, LOG_PAGE, offset);
        this.ctx.storage.transactionSync(() => this.index.upsert(commits.map(toIndexed), rebuild.gen));
        offset += commits.length;
        done = commits.length < LOG_PAGE || offset >= MAX_INDEXED_COMMITS;
      }
    });
    if (!done) {
      await save({ ...state, rebuild: { ...rebuild, offset } });
      return true;
    }
    this.ctx.storage.transactionSync(() => this.index.dropOtherGenerations(rebuild.gen));
    await save({
      ...state,
      head: rebuild.target,
      gen: rebuild.gen,
      rebuild: null,
      schema: INDEX_SCHEMA,
      truncated: offset >= MAX_INDEXED_COMMITS,
    });
    return this.publish(state.repoId);
  }

  private async publish(repoId: string): Promise<boolean> {
    const rows = this.index.pendingPublish(PUBLISH_CHUNK);
    if (rows.length === 0) return false;
    await publishRows(repoId, rows);
    this.ctx.storage.transactionSync(() => this.index.markPublished(rows));
    return rows.length === PUBLISH_CHUNK;
  }
}

export function repoInsights(repoId: string) {
  return env.REPO_INSIGHTS.get(env.REPO_INSIGHTS.idFromName(repoId));
}

/** Never fails the caller: the index self-heals on the next kick or Insights visit. */
export async function kickRepoInsights(repoId: string): Promise<void> {
  try {
    await repoInsights(repoId).kick(repoId);
  } catch (error) {
    console.error("Repo insights kick failed", repoId, error);
  }
}