import { WorkflowEntrypoint, type WorkflowEvent, type WorkflowStep, type WorkflowStepConfig } from 'cloudflare:workers';
import {
  applyChanges,
  countUrls,
  getState,
  markStale,
  markSynced,
  putState,
  saveAccounts,
  saveDeclarations,
  saveHandles,
  staleOwners,
} from './db';
import { COLLECTION_SPECS, DECLARATION, extractDeclaration, extractUrlAction } from './hydrant';
import { hydrantRepos, type RepoHandle, type Repos } from './identity';
import { readJetstream, subscribeJetstream, type Subscribe } from './jetstream';
import { getStaticUrls, renderShards, resetSitemap } from './sitemap';
import type { Env } from './types';
import { PROTOCOL_MAX_BYTES, PROTOCOL_MAX_URLS } from './xml';

const DEFAULT_HYDRANT_URL = 'https://index.tangled.network';

function hydrantUrl(env: Env): string {
  return (env.HYDRANT_URL || DEFAULT_HYDRANT_URL).replace(/\/+$/, '');
}

export interface SyncDeps {
  subscribe: Subscribe;
  repos: (env: Env) => Repos;
  idleMs: number;
  closeMs: number;
}

const defaultDeps: SyncDeps = {
  subscribe: subscribeJetstream,
  // the vpc service only takes http and routes by itself, so the host there means nothing
  repos: (env) => (env.HYDRANT ? hydrantRepos(env.HYDRANT, 'http://hydrant') : hydrantRepos(globalThis, hydrantUrl(env))),
  idleMs: 3000,
  closeMs: 10_000,
};

export function parseConfigLimits(env: Env): { maxUrls: number; maxBytes: number } {
  const parsedUrls = parseInt(env.SHARD_MAX_URLS || '', 10);
  const parsedBytes = parseInt(env.SHARD_MAX_BYTES || '', 10);

  const maxUrls = !isNaN(parsedUrls) && parsedUrls > 0
    ? Math.min(parsedUrls, PROTOCOL_MAX_URLS)
    : PROTOCOL_MAX_URLS;

  const maxBytes = !isNaN(parsedBytes) && parsedBytes > 0
    ? Math.min(parsedBytes, PROTOCOL_MAX_BYTES)
    : PROTOCOL_MAX_BYTES;

  return { maxUrls, maxBytes };
}

export default {
  async fetch(request: Request, env: Env, ctx: ExecutionContext): Promise<Response> {
    const url = new URL(request.url);
    const pathname = url.pathname;

    if (pathname === '/sitemap.xml') {
      const obj = await env.SITEMAP_BUCKET.get('sitemap.xml');
      if (!obj) {
        return new Response('Sitemap index not found. Has sync run yet?', { status: 404 });
      }

      return new Response(obj.body, {
        headers: {
          'Content-Type': 'application/xml; charset=utf-8',
          'Cache-Control': 'public, max-age=86400, s-maxage=86400',
        },
      });
    }

    if (pathname.startsWith('/sitemaps/')) {
      const filename = pathname.replace(/^\/sitemaps\//, '');
      const obj = await env.SITEMAP_BUCKET.get(`sitemaps/${filename}`);
      if (!obj) {
        return new Response('Sitemap shard not found', { status: 404 });
      }

      return new Response(obj.body, {
        headers: {
          'Content-Type': 'application/gzip',
          'Cache-Control': 'public, max-age=3600, stale-while-revalidate=86400',
        },
      });
    }



    return new Response('Not Found', { status: 404 });
  },
};

export interface Steps {
  do<T>(name: string, fn: () => Promise<T>): Promise<T>;
}

// the workflow checkpoints every step, tests and local runs just call them
const inline: Steps = { do: (_name, fn) => fn() };

export interface SyncOptions {
  reset?: boolean;
  startedAt?: Date;
}

export interface SyncResult {
  syncedEvents: number;
  mutatedUrls: number;
  walkedRepos: number;
  lookedUpHandles: number;
  newCursor: string;
  totalUrls: number;
}

const MAX_PAGES = 10;

const REPO_PAGE = 1000;
const WALK_PAGES = 10;
const MAX_WALK_STEPS = 100;

// a page of reads that all time out still fits well inside a step
const LOOKUP_PAGE = 200;
const LOOKUP_CONCURRENCY = 8;
const MAX_LOOKUP_STEPS = 10;

// the first run, and the first after a reset, pages through every handle hydrant has
// instead of reading owners one by one. a walk that gets cut off carries on next run
async function walkRepos(
  db: D1Database,
  repos: Repos,
  now: Date,
): Promise<{ repos: number; done: boolean; failed: boolean }> {
  if (await getState(db, 'handles_walked')) return { repos: 0, done: true, failed: false };
  let after = await getState(db, 'handles_after');
  let count = 0;
  try {
    for (let page = 0; page < WALK_PAGES; page++) {
      const found = await repos.list(after, REPO_PAGE);
      await saveHandles(db, found);
      count += found.length;
      if (found.length < REPO_PAGE) {
        await putState(db, 'handles_walked', now.toISOString());
        return { repos: count, done: true, failed: false };
      }
      after = found[found.length - 1].did;
      await putState(db, 'handles_after', after);
    }
  } catch (err) {
    // the sitemap still goes out, just with dids where handles are missing
    console.warn(`[sitemap] couldn't page through hydrant's repos: ${err}`);
    return { repos: count, done: false, failed: true };
  }
  return { repos: count, done: false, failed: false };
}

// owners new since the walk or touched by an identity event. a failed read stays
// stale for the next run
async function lookUpHandles(
  db: D1Database,
  repos: Repos,
  after: string,
): Promise<{ looked: number; failed: number; last: string }> {
  const dids = await staleOwners(db, after, LOOKUP_PAGE);
  const found: RepoHandle[] = [];
  let failed = 0;
  let next = 0;
  const worker = async () => {
    while (next < dids.length) {
      const did = dids[next++];
      try {
        found.push({ did, handle: await repos.get(did) });
      } catch {
        failed++;
      }
    }
  };
  await Promise.all(Array.from({ length: LOOKUP_CONCURRENCY }, worker));
  await saveHandles(db, found);
  if (failed > 0) console.warn(`[sitemap] ${failed} of ${dids.length} handle reads failed`);
  return { looked: dids.length, failed, last: dids.at(-1) ?? after };
}

export async function syncFromHydrant(
  env: Env,
  deps: SyncDeps = defaultDeps,
  steps: Steps = inline,
  options: SyncOptions = {},
): Promise<SyncResult> {
  const db = env.SITEMAP_DB;
  const baseUrl = env.BASE_URL || 'https://tangled.org';
  const limits = parseConfigLimits(env);
  // a workflow replays this on resume, so the time has to come from the instance
  const now = options.startedAt ?? new Date();
  const parsedBatch = parseInt(env.SYNC_BATCH_SIZE || '', 10);
  const batchSize = !isNaN(parsedBatch) && parsedBatch > 0 ? parsedBatch : 10000;
  const seen = new Set<string>();
  let syncedEvents = 0;
  let mutatedUrls = 0;

  if (options.reset) {
    await steps.do('reset', () => resetSitemap(db, env.SITEMAP_BUCKET));
  }

  mutatedUrls += await steps.do('seed', async () => {
    if ((await countUrls(db)) > 0) return 0;
    const seeds = getStaticUrls(baseUrl, now).map((entry) => ({ type: 'upsert' as const, url: entry.loc, entry }));
    return applyChanges(db, seeds, limits, undefined, now);
  });

  for (let page = 1; page <= MAX_PAGES; page++) {
    const batch = await steps.do(`batch ${page}`, async () => {
      // read here rather than carried over, so a retried step starts from what landed
      const cursor = Number(await getState(db, 'cursor')) || 0;
      const read = await readJetstream(
        {
          url: hydrantUrl(env),
          cursor,
          collections: [...Object.keys(COLLECTION_SPECS), DECLARATION],
          maxEvents: batchSize,
          idleMs: deps.idleMs,
          closeMs: deps.closeMs,
          untilUs: now.getTime() * 1000,
          maxReconnects: 3,
          seen,
        },
        deps.subscribe,
      );
      if (read.events.length === 0) {
        if (read.error) throw new Error(read.error);
        return { events: 0, mutations: 0, more: false };
      }
      const changes = read.events.flatMap((evt) => extractUrlAction(evt, baseUrl, now) ?? []);
      // before the cursor moves, so a retried batch can't skip it
      await markStale(db, read.events.flatMap((evt) => (evt.kind === 'identity' ? [evt.did] : [])));
      await saveAccounts(db, read.events.flatMap((evt) => (evt.kind === 'account' ? [evt.account] : [])));
      await saveDeclarations(db, read.events.flatMap((evt) => extractDeclaration(evt) ?? []));
      const mutations = await applyChanges(db, changes, limits, String(read.cursor), now);
      return { events: read.events.length, mutations, more: read.full && !read.error };
    });
    syncedEvents += batch.events;
    mutatedUrls += batch.mutations;
    if (!batch.more) break;
  }

  const repos = deps.repos(env);
  let walkedRepos = 0;
  let walked = false;
  for (let step = 1; step <= MAX_WALK_STEPS && !walked; step++) {
    const res = await steps.do(`walk ${step}`, () => walkRepos(db, repos, now));
    walkedRepos += res.repos;
    walked = res.done;
    if (res.failed) break;
  }

  // until the walk is through, most owners are still in it
  let lookedUpHandles = 0;
  let after = '';
  for (let step = 1; walked && step <= MAX_LOOKUP_STEPS; step++) {
    const res = await steps.do(`lookups ${step}`, () => lookUpHandles(db, repos, after));
    lookedUpHandles += res.looked;
    // nothing answering means hydrant is down, so render with what we have
    if (res.looked < LOOKUP_PAGE || res.failed === res.looked) break;
    after = res.last;
  }

  await steps.do('render', () => renderShards(db, env.SITEMAP_BUCKET, baseUrl, now));

  return steps.do('finish', async () => {
    if (syncedEvents > 0) await markSynced(db, now);
    return {
      syncedEvents,
      mutatedUrls,
      walkedRepos,
      lookedUpHandles,
      newCursor: (await getState(db, 'cursor')) ?? '0',
      totalUrls: await countUrls(db),
    };
  });
}

const STEP_CONFIG: WorkflowStepConfig = {
  retries: { limit: 3, delay: '30 seconds', backoff: 'exponential' },
  timeout: '15 minutes',
};

// started by its own schedule at midnight, or by hand with
// wrangler workflows trigger <name> '{"reset":true}'
export class SyncWorkflow extends WorkflowEntrypoint<Env, SyncOptions> {
  async run(event: Readonly<WorkflowEvent<SyncOptions>>, step: WorkflowStep): Promise<SyncResult> {
    const steps: Steps = { do: (name, fn) => step.do(name, STEP_CONFIG, fn as () => Promise<never>) };
    const res = await syncFromHydrant(this.env, defaultDeps, steps, {
      reset: event.payload?.reset,
      startedAt: event.timestamp,
    });
    console.log(`[sitemap] sync finished: ${JSON.stringify(res)}`);
    return res;
  }
}