import { describe, expect, it } from 'vitest';
import type { JetstreamEvent } from '@atcute/jetstream';
import { parseConfigLimits, syncFromHydrant, type SyncDeps } from '../src/index';
import type { Repos } from '../src/identity';
import type { Env } from '../src/types';
import { TestD1 } from './d1';
import { account, commit, deadJetstream, declaration, fakeJetstream, identity, manifest } from './events';
import { buildUrlsetXml, gzipDecompress, measureXmlBytes } from '../src/xml';

function repo(
  seconds: number,
  owner: string,
  key: string,
  operation: 'create' | 'update' | 'delete' = 'create',
  record: Record<string, unknown> = {},
): JetstreamEvent {
  const repoDid = `${owner}-${key}`;
  if (operation === 'delete') return commit(seconds, repoDid, 'org.tangled.repo.manifest', 'self', 'delete');
  const { name = key, ...rest } = record;
  return manifest(seconds, repoDid, owner, name as string, operation, rest);
}

// claims the lowercased name, the same key the web uses
function claimed(events: JetstreamEvent[]): JetstreamEvent[] {
  return events.flatMap((evt) => {
    if (evt.kind !== 'commit' || evt.commit.collection !== 'org.tangled.repo.manifest' || evt.commit.operation === 'delete') {
      return [evt];
    }
    const { owner, name } = evt.commit.record.declaration as { owner: string; name: string };
    return [declaration(evt.time_us / 1_000_000, owner, name.toLowerCase(), evt.did), evt];
  });
}

function claimedJetstream(source: () => JetstreamEvent[]) {
  return fakeJetstream(() => claimed(source()));
}

class MockR2Object {
  constructor(private buffer: ArrayBuffer, public httpMetadata?: any) {}
  async arrayBuffer(): Promise<ArrayBuffer> {
    return this.buffer;
  }
}

class MockR2Bucket {
  store = new Map<string, { buffer: ArrayBuffer; httpMetadata?: any }>();
  putCount = 0;

  async get(key: string): Promise<MockR2Object | null> {
    const item = this.store.get(key);
    if (!item) return null;
    return new MockR2Object(item.buffer, item.httpMetadata);
  }

  async put(
    key: string,
    value: ArrayBuffer | Uint8Array | string,
    options?: { httpMetadata?: any },
  ): Promise<void> {
    this.putCount++;
    let buffer: ArrayBuffer;
    if (typeof value === 'string') {
      buffer = new TextEncoder().encode(value).buffer as ArrayBuffer;
    } else if (value instanceof Uint8Array) {
      buffer = value.buffer.slice(value.byteOffset, value.byteOffset + value.byteLength) as ArrayBuffer;
    } else {
      buffer = value;
    }
    this.store.set(key, { buffer, httpMetadata: options?.httpMetadata });
  }

  async list(): Promise<{ objects: { key: string }[]; truncated: false }> {
    return { objects: [...this.store.keys()].map((key) => ({ key })), truncated: false };
  }

  async delete(keys: string[]): Promise<void> {
    for (const key of keys) this.store.delete(key);
  }
}

// hydrant's repos as the tunnel serves them, where nobody has a handle unless a test
// gives them one. changing `handles` later is someone changing theirs
function hydrant(handles: Record<string, string | null> = {}): Repos & { asked: string[]; pages: (string | undefined)[] } {
  const fake = {
    asked: [] as string[],
    pages: [] as (string | undefined)[],
    async list(after: string | undefined, limit: number) {
      fake.pages.push(after);
      return Object.keys(handles)
        .sort()
        .filter((did) => after === undefined || did > after)
        .slice(0, limit)
        .map((did) => ({ did, handle: handles[did] }));
    },
    async get(did: string) {
      fake.asked.push(did);
      return handles[did] ?? null;
    },
  };
  return fake;
}

const down: Repos = {
  async list() {
    throw new Error('hydrant is down');
  },
  async get() {
    throw new Error('hydrant is down');
  },
};

function deps(subscribe: SyncDeps['subscribe'], repos: Repos = hydrant()): SyncDeps {
  return { subscribe, repos: () => repos, idleMs: 5, closeMs: 5 };
}

function users(count: number): Record<string, string> {
  return Object.fromEntries(
    Array.from({ length: count }, (_, i) => [`did:plc:user${String(i).padStart(4, '0')}`, `user${i}.test`]),
  );
}

async function shardXml(bucket: MockR2Bucket, shard: number): Promise<string> {
  const obj = await bucket.get(`sitemaps/sitemap-${String(shard).padStart(4, '0')}.xml.gz`);
  return gzipDecompress(await obj!.arrayBuffer());
}

function handleRow(db: TestD1, did: string): { handle: string | null; stale: number } | undefined {
  return db.query<{ handle: string | null; stale: number }>('SELECT handle, stale FROM handles WHERE did = ?', did)[0];
}

function shardOf(db: TestD1, url: string): number | undefined {
  return db.query<{ shard: number }>('SELECT shard FROM urls WHERE url = ?', url)[0]?.shard;
}

function recordUrl(db: TestD1, uri: string): string | undefined {
  return db.query<{ url: string }>('SELECT url FROM records WHERE uri = ?', uri)[0]?.url;
}

function shardCounts(db: TestD1): number[] {
  return db.query<{ n: number }>('SELECT count(*) AS n FROM urls GROUP BY shard ORDER BY shard').map((row) => row.n);
}

describe('Full syncFromHydrant workflow with Sharding & Multi-Sync', () => {
  it('handles multi-sync, updates historical shards in-place, and enforces caps', async () => {
    const db = new TestD1();
    const bucket = new MockR2Bucket();

    // ==========================================
    // SYNC 1: Initial creation + multi-shard rollover
    // ==========================================
    const sync1Events: JetstreamEvent[] = [
      identity(1, 'did:plc:dawn', 'dawn.tngl.sh'),
      commit(2, 'did:plc:dawn', 'org.tangled.actor.profile', 'self', 'create', { createdAt: '2026-08-01T00:00:00Z' }),
    ];

    // 58 repo events -> 61 total URLs (2 static + 1 profile + 58 repos)
    for (let i = 1; i <= 58; i++) {
      sync1Events.push(
        repo(10 + i, 'did:plc:dawn', `repo-${i}`, 'create', {
          name: `repo-${i}`,
          createdAt: '2026-08-15T00:00:00Z',
        }),
      );
    }

    let currentMockEvents = sync1Events;
    const jetstream = claimedJetstream(() => currentMockEvents);

    const env: Env = {
      SITEMAP_DB: db.asD1(),
      SITEMAP_BUCKET: bucket as any,
      BASE_URL: 'https://tangled.org',
      SHARD_MAX_URLS: '25', // 25 max per shard
    };

    const res1 = await syncFromHydrant(env, deps(jetstream.subscribe));
    // 118 is two events per repo (manifest and declaration), plus identity and profile
    expect(res1.syncedEvents).toBe(118);
    expect(res1.totalUrls).toBe(61);
    expect(res1.newCursor).toBe('68000000');

    expect(shardCounts(db)).toEqual([25, 25, 11]);
    expect(shardOf(db, 'https://tangled.org/did:plc:dawn/repo-1')).toBe(1);

    // ==========================================
    // SYNC 2:
    // - Asserts in-place update in historical shard 1 (repo-2)
    // - Asserts delete against historical sealed shard 1 (repo-1)
    // ==========================================
    currentMockEvents = [
      repo(100, 'did:plc:dawn', 'new-tool', 'create', { createdAt: '2026-09-01T00:00:00Z' }),
      repo(101, 'did:plc:dawn', 'repo-1', 'delete'),
      repo(102, 'did:plc:dawn', 'repo-2', 'update', { createdAt: '2026-09-05T00:00:00Z' }),
    ];
    const r2PutsBeforeSync2 = bucket.putCount;
    const res2 = await syncFromHydrant(env, deps(jetstream.subscribe));
    expect(res2.syncedEvents).toBe(5);
    expect(res2.mutatedUrls).toBe(3);
    expect(res2.newCursor).toBe('102000000');
    expect(res2.totalUrls).toBe(61); // -1 deleted, +1 in-place updated, +1 new in active shard

    // shard 1 for the update and delete, shard 3 for new-tool, and the index
    expect(bucket.putCount - r2PutsBeforeSync2).toBe(3);

    // the second run picked up 10s before where the first one stopped
    expect(jetstream.cursors).toEqual([0, 58000000]);

    expect(shardOf(db, 'https://tangled.org/did:plc:dawn/new-tool')).toBe(3);
    // repo-2 is updated where it is, in the full shard 1
    expect(shardOf(db, 'https://tangled.org/did:plc:dawn/repo-2')).toBe(1);
    expect(shardOf(db, 'https://tangled.org/did:plc:dawn/repo-1')).toBeUndefined();
    expect(shardCounts(db)).toEqual([24, 25, 12]);

    const shard1Obj = await bucket.get('sitemaps/sitemap-0001.xml.gz');
    const shard1Xml = await gzipDecompress(await shard1Obj!.arrayBuffer());
    expect(shard1Xml.match(/<url>/g)).toHaveLength(24);
    expect(shard1Xml).not.toContain('<loc>https://tangled.org/did:plc:dawn/repo-1</loc>');
    expect(shard1Xml).toContain('<loc>https://tangled.org/did:plc:dawn/repo-2</loc>');
    expect(shard1Xml).toContain('<lastmod>2026-09-05</lastmod>');

    // ==========================================
    // SYNC 3: a sync with nothing new writes nothing
    // ==========================================
    const r2PutsBefore = bucket.putCount;
    const dbWritesBefore = db.writes;

    currentMockEvents = [];
    const res3 = await syncFromHydrant(env, deps(jetstream.subscribe));
    expect(res3.syncedEvents).toBe(0);
    expect(res3.mutatedUrls).toBe(0);

    expect(bucket.putCount).toBe(r2PutsBefore);
    expect(db.writes).toBe(dbWritesBefore);

    // ==========================================
    // SYNC 4: Connection failure propagation
    // ==========================================
    // throwing is what makes the workflow retry the step
    await expect(syncFromHydrant(env, deps(deadJetstream))).rejects.toThrow('disconnected');
    expect(db.query<{ value: string }>("SELECT value FROM state WHERE key = 'cursor'")[0].value).toBe('102000000');
  });

  it('finds deleted and renamed repos through their record uri', async () => {
    const db = new TestD1();
    const bucket = new MockR2Bucket();
    let events = [
      repo(1, 'did:plc:alice', '3mtid1', 'create', { name: 'old-name' }),
      repo(2, 'did:plc:alice', '3mtid2', 'create', { name: 'doomed' }),
    ];
    const jetstream = claimedJetstream(() => events);
    const env: Env = { SITEMAP_DB: db.asD1(), SITEMAP_BUCKET: bucket as any, BASE_URL: 'https://tangled.org' };

    await syncFromHydrant(env, deps(jetstream.subscribe));
    expect(recordUrl(db, 'at://did:plc:alice-3mtid2/org.tangled.repo.manifest/self')).toBe('https://tangled.org/did:plc:alice/doomed');

    events = [
      repo(20, 'did:plc:alice', '3mtid1', 'update', { name: 'new-name' }),
      repo(21, 'did:plc:alice', '3mtid2', 'delete'),
    ];
    await syncFromHydrant(env, deps(jetstream.subscribe));

    const shard = await gzipDecompress(await (await bucket.get('sitemaps/sitemap-0001.xml.gz'))!.arrayBuffer());
    expect(shard).toContain('<loc>https://tangled.org/did:plc:alice/new-name</loc>');
    expect(shard).not.toContain('old-name');
    expect(shard).not.toContain('doomed');
    expect(shardOf(db, 'https://tangled.org/did:plc:alice/old-name')).toBeUndefined();
    expect(shardOf(db, 'https://tangled.org/did:plc:alice/doomed')).toBeUndefined();
    expect(recordUrl(db, 'at://did:plc:alice-3mtid2/org.tangled.repo.manifest/self')).toBeUndefined();
  });

  it('moves a repo whose manifest names a new owner', async () => {
    const db = new TestD1();
    const bucket = new MockR2Bucket();
    let events = [manifest(1, 'did:plc:repo1', 'did:plc:alice', 'tool')];
    const jetstream = claimedJetstream(() => events);
    const repos = hydrant({ 'did:plc:alice': 'alice.test', 'did:plc:org': 'org.test' });
    const env: Env = { SITEMAP_DB: db.asD1(), SITEMAP_BUCKET: bucket as any, BASE_URL: 'https://tangled.org' };

    await syncFromHydrant(env, deps(jetstream.subscribe, repos));
    expect(await shardXml(bucket, 1)).toContain('<loc>https://tangled.org/alice.test/tool</loc>');

    events = [manifest(20, 'did:plc:repo1', 'did:plc:org', 'tool', 'update')];
    await syncFromHydrant(env, deps(jetstream.subscribe, repos));
    const xml = await shardXml(bucket, 1);
    expect(xml).toContain('<loc>https://tangled.org/org.test/tool</loc>');
    expect(xml).not.toContain('alice');
    expect(recordUrl(db, 'at://did:plc:repo1/org.tangled.repo.manifest/self')).toBe('https://tangled.org/did:plc:org/tool');
  });

  it('leaves out what deleted, deactivated and taken down accounts hold, until they come back', async () => {
    const db = new TestD1();
    const bucket = new MockR2Bucket();
    let events = [
      commit(1, 'did:plc:alice', 'org.tangled.actor.profile', 'self'),
      commit(2, 'did:plc:alice', 'org.tangled.snippet.string', '3mtid1'),
      manifest(3, 'did:plc:repo1', 'did:plc:alice', 'tool'),
      manifest(4, 'did:plc:repo2', 'did:plc:bob', 'gone'),
      manifest(5, 'did:plc:repo3', 'did:plc:bob', 'kept'),
      account(6, 'did:plc:alice', false, 'takendown'),
      account(7, 'did:plc:repo2', false, 'deleted'),
      account(8, 'did:plc:bob', true),
    ];
    const jetstream = claimedJetstream(() => events);
    const env: Env = { SITEMAP_DB: db.asD1(), SITEMAP_BUCKET: bucket as any, BASE_URL: 'https://tangled.org' };

    await syncFromHydrant(env, deps(jetstream.subscribe));
    let xml = await shardXml(bucket, 1);
    expect(xml).not.toContain('did:plc:alice');
    expect(xml).not.toContain('gone');
    expect(xml).toContain('<loc>https://tangled.org/did:plc:bob/kept</loc>');

    events = [account(20, 'did:plc:alice', true)];
    await syncFromHydrant(env, deps(jetstream.subscribe));
    xml = await shardXml(bucket, 1);
    expect(xml).toContain('<loc>https://tangled.org/did:plc:alice</loc>');
    expect(xml).toContain('<loc>https://tangled.org/strings/did:plc:alice/3mtid1</loc>');
    expect(xml).toContain('<loc>https://tangled.org/did:plc:alice/tool</loc>');
    expect(xml).not.toContain('gone');
  });

  it('rerenders nothing for an account status it already had', async () => {
    const db = new TestD1();
    const bucket = new MockR2Bucket();
    let events = [manifest(1, 'did:plc:repo1', 'did:plc:alice', 'tool'), account(2, 'did:plc:repo1', false, 'deleted')];
    const jetstream = claimedJetstream(() => events);
    const env: Env = { SITEMAP_DB: db.asD1(), SITEMAP_BUCKET: bucket as any, BASE_URL: 'https://tangled.org' };
    await syncFromHydrant(env, deps(jetstream.subscribe));
    expect(await shardXml(bucket, 1)).not.toContain('tool');

    events = [account(20, 'did:plc:repo1', false, 'deleted'), account(21, 'did:plc:carol', true)];
    const puts = bucket.putCount;
    await syncFromHydrant(env, deps(jetstream.subscribe));
    expect(bucket.putCount).toBe(puts);
  });

  it("shows a repo only while its owner's declaration claims it", async () => {
    const db = new TestD1();
    const bucket = new MockR2Bucket();
    let events = [manifest(1, 'did:plc:repo1', 'did:plc:alice', 'Tool'), manifest(2, 'did:plc:repo2', 'did:plc:alice', 'other')];
    const jetstream = fakeJetstream(() => events);
    const env: Env = { SITEMAP_DB: db.asD1(), SITEMAP_BUCKET: bucket as any, BASE_URL: 'https://tangled.org' };

    await syncFromHydrant(env, deps(jetstream.subscribe));
    expect(await shardXml(bucket, 1)).not.toContain('Tool');

    events = [declaration(10, 'did:plc:alice', 'tool', 'did:plc:repo1'), declaration(11, 'did:plc:alice', 'other', 'did:plc:someone-else')];
    await syncFromHydrant(env, deps(jetstream.subscribe));
    let xml = await shardXml(bucket, 1);
    expect(xml).toContain('<loc>https://tangled.org/did:plc:alice/Tool</loc>');
    expect(xml).not.toContain('other');

    events = [commit(20, 'did:plc:alice', 'org.tangled.repo.declaration', 'tool', 'delete')];
    await syncFromHydrant(env, deps(jetstream.subscribe));
    xml = await shardXml(bucket, 1);
    expect(xml).not.toContain('Tool');
  });

  it('paginates across multiple batches to catch up in a single sync run', async () => {
    const db = new TestD1();
    const bucket = new MockR2Bucket();

    const allEvents = [
      identity(0, 'did:plc:multi', 'multi-user'),
      ...Array.from({ length: 15 }, (_, i) =>
        repo(i + 1, 'did:plc:multi', `repo-${i + 1}`, 'create', {
          createdAt: '2026-08-01T00:00:00Z',
        }),
      ),
    ];
    const jetstream = claimedJetstream(() => allEvents);

    const env: Env = {
      SITEMAP_DB: db.asD1(),
      SITEMAP_BUCKET: bucket as any,
      BASE_URL: 'https://tangled.org',
      SYNC_BATCH_SIZE: '4',
    };

    const res = await syncFromHydrant(env, deps(jetstream.subscribe));
    expect(res.syncedEvents).toBe(31);
    expect(res.newCursor).toBe('15000000');
    expect(jetstream.cursors).toHaveLength(8); // 7 full pages of 4, then a quiet one

    for (let i = 1; i <= 15; i++) {
      expect(shardOf(db, `https://tangled.org/did:plc:multi/repo-${i}`)).toBe(1);
    }
  });

  it('enforces byte-size cap rollover', async () => {
    const db = new TestD1();
    const bucket = new MockR2Bucket();
    const jetstream = claimedJetstream(() => [
      identity(1, 'did:plc:alice', 'alice'),
      repo(2, 'did:plc:alice', 'long-repository-name-to-consume-bytes'),
    ]);

    // Tiny 350-byte max cap -> will force rollover after ~2 URLs
    const env: Env = {
      SITEMAP_DB: db.asD1(),
      SITEMAP_BUCKET: bucket as any,
      BASE_URL: 'https://tangled.org',
      SHARD_MAX_BYTES: '350', // very low byte limit
    };

    const res = await syncFromHydrant(env, deps(jetstream.subscribe));

    expect(shardCounts(db).length).toBeGreaterThanOrEqual(2);
    for (const { shard } of db.query<{ shard: number }>('SELECT DISTINCT shard FROM urls')) {
      const obj = await bucket.get(`sitemaps/sitemap-${String(shard).padStart(4, '0')}.xml.gz`);
      expect(new TextEncoder().encode(await gzipDecompress(await obj!.arrayBuffer())).byteLength).toBeLessThanOrEqual(350);
    }
  });

  it('starts over from an empty sitemap on reset', async () => {
    const db = new TestD1();
    const bucket = new MockR2Bucket();
    let events = [1, 2, 3].map((i) => repo(i, 'did:plc:alice', `old-repo-${i}`));
    const jetstream = claimedJetstream(() => events);
    const env: Env = { SITEMAP_DB: db.asD1(), SITEMAP_BUCKET: bucket as any, BASE_URL: 'https://tangled.org', SHARD_MAX_URLS: '2' };

    await syncFromHydrant(env, deps(jetstream.subscribe));
    expect(bucket.store.has('sitemaps/sitemap-0003.xml.gz')).toBe(true);

    events = [repo(5, 'did:plc:alice', 'new-repo')];
    const res = await syncFromHydrant(env, deps(jetstream.subscribe), undefined, { reset: true });
    expect(jetstream.cursors.at(-1)).toBe(0);
    expect(res.totalUrls).toBe(3);
    expect(db.query("SELECT url FROM urls WHERE url LIKE '%old-repo%'")).toEqual([]);
    expect(shardOf(db, 'https://tangled.org/did:plc:alice/new-repo')).toBe(2);
    expect([...bucket.store.keys()].sort()).toEqual(['sitemap.xml', 'sitemaps/sitemap-0001.xml.gz', 'sitemaps/sitemap-0002.xml.gz']);
  });

  it('validates config limits and handles NaN safely', () => {
    expect(parseConfigLimits({ SITEMAP_DB: {} as any, SITEMAP_BUCKET: {} as any, SHARD_MAX_URLS: 'abc' })).toEqual({
      maxUrls: 50000,
      maxBytes: 52428800,
    });
    expect(parseConfigLimits({ SITEMAP_DB: {} as any, SITEMAP_BUCKET: {} as any, SHARD_MAX_URLS: '-5' })).toEqual({
      maxUrls: 50000,
      maxBytes: 52428800,
    });
    expect(parseConfigLimits({ SITEMAP_DB: {} as any, SITEMAP_BUCKET: {} as any, SHARD_MAX_URLS: '1000' })).toEqual({
      maxUrls: 1000,
      maxBytes: 52428800,
    });
  });
});

const EMPTY_SHARD = measureXmlBytes(buildUrlsetXml([]));

// what `applyChanges` thinks a shard weighs has to match what it renders to
async function expectSizesTracked(db: TestD1, bucket: MockR2Bucket): Promise<void> {
  for (const { shard, bytes } of db.query<{ shard: number; bytes: number }>(
    'SELECT shard, sum(bytes) AS bytes FROM urls GROUP BY shard',
  )) {
    // `urlBytes` counts a joining newline for every url, and there's one fewer
    expect(measureXmlBytes(await shardXml(bucket, shard))).toBe(EMPTY_SHARD + bytes - 1);
  }
}

describe('handles', () => {
  const day = 24 * 60 * 60 * 1000;
  const startedAt = new Date('2026-10-01T00:00:00Z');
  const env = (db: TestD1, bucket: MockR2Bucket, vars: Partial<Env> = {}): Env => ({
    SITEMAP_DB: db.asD1(),
    SITEMAP_BUCKET: bucket as any,
    BASE_URL: 'https://tangled.org',
    ...vars,
  });

  it('renders owners as their handles and keeps the did when there is none', async () => {
    const db = new TestD1();
    const bucket = new MockR2Bucket();
    const jetstream = claimedJetstream(() => [
      commit(1, 'did:plc:alice', 'org.tangled.actor.profile', 'self'),
      repo(2, 'did:plc:alice', '3mtid1', 'create', { name: 'tool' }),
      commit(3, 'did:plc:alice', 'org.tangled.snippet.string', '3mtid2'),
      repo(4, 'did:plc:bob', 'thing'),
    ]);
    const repos = hydrant({ 'did:plc:alice': 'alice.test' });

    const res = await syncFromHydrant(env(db, bucket), deps(jetstream.subscribe, repos));
    expect(res.walkedRepos).toBe(1);
    // the walk had alice, bob is the one it didn't know
    expect(repos.asked).toEqual(['did:plc:bob']);
    expect(res.lookedUpHandles).toBe(1);

    const xml = await shardXml(bucket, 1);
    expect(xml).toContain('<loc>https://tangled.org/alice.test</loc>');
    expect(xml).toContain('<loc>https://tangled.org/alice.test/tool</loc>');
    expect(xml).toContain('<loc>https://tangled.org/strings/alice.test/3mtid2</loc>');
    expect(xml).toContain('<loc>https://tangled.org/did:plc:bob/thing</loc>');
    expect(xml).not.toContain('did:plc:alice');
    // stored by did, so a handle change never moves them
    expect(shardOf(db, 'https://tangled.org/did:plc:alice/tool')).toBe(1);
    await expectSizesTracked(db, bucket);
  });

  it('walks every repo once, and again after a reset', async () => {
    const db = new TestD1();
    const bucket = new MockR2Bucket();
    const handles = users(2500);
    const jetstream = claimedJetstream(() => [
      repo(1, 'did:plc:user0042', 'tool'),
      repo(2, 'did:plc:user2499', 'tool'),
    ]);
    const repos = hydrant(handles);

    const res = await syncFromHydrant(env(db, bucket), deps(jetstream.subscribe, repos));
    expect(res.walkedRepos).toBe(2500);
    expect(repos.pages).toEqual([undefined, 'did:plc:user0999', 'did:plc:user1999']);
    expect(repos.asked).toEqual([]);
    expect(await shardXml(bucket, 1)).toContain('<loc>https://tangled.org/user2499.test/tool</loc>');

    const again = await syncFromHydrant(env(db, bucket), deps(jetstream.subscribe, repos));
    expect(again.walkedRepos).toBe(0);
    expect(repos.pages).toHaveLength(3);

    await syncFromHydrant(env(db, bucket), deps(jetstream.subscribe, repos), undefined, { reset: true });
    expect(repos.pages).toHaveLength(6);
    expect(repos.asked).toEqual([]);
  });

  it('carries on a walk that broke off from where it got to', async () => {
    const db = new TestD1();
    const bucket = new MockR2Bucket();
    const jetstream = claimedJetstream(() => [repo(1, 'did:plc:user2000', 'tool')]);
    const repos = hydrant(users(2500));
    let calls = 0;
    const flaky: Repos = {
      list: (after, limit) => (++calls === 2 ? down.list(after, limit) : repos.list(after, limit)),
      get: (did) => repos.get(did),
    };

    const res = await syncFromHydrant(env(db, bucket), deps(jetstream.subscribe, flaky));
    expect(res.walkedRepos).toBe(1000);
    // the rest of the walk will have most owners, so nobody gets read one by one yet
    expect(res.lookedUpHandles).toBe(0);
    expect(await shardXml(bucket, 1)).toContain('<loc>https://tangled.org/did:plc:user2000/tool</loc>');

    const next = await syncFromHydrant(env(db, bucket), deps(jetstream.subscribe, flaky));
    expect(next.walkedRepos).toBe(1500);
    expect(repos.pages).toEqual([undefined, 'did:plc:user0999', 'did:plc:user1999']);
    expect(db.query('SELECT did FROM handles')).toHaveLength(2500);
    expect(await shardXml(bucket, 1)).toContain('<loc>https://tangled.org/user2000.test/tool</loc>');
  });

  it('rerenders only the shards of an owner whose handle changed', async () => {
    const db = new TestD1();
    const bucket = new MockR2Bucket();
    let events = [
      repo(1, 'did:plc:alice', 'alice-1'),
      repo(2, 'did:plc:bob', 'bob-1'),
      repo(3, 'did:plc:bob', 'bob-2'),
      repo(4, 'did:plc:bob', 'bob-3'),
      repo(5, 'did:plc:alice', 'alice-2'),
    ];
    const jetstream = claimedJetstream(() => events);
    const vars = { SHARD_MAX_URLS: '3' };
    const handles: Record<string, string> = { 'did:plc:alice': 'alice.test', 'did:plc:bob': 'bob.test' };
    const repos = hydrant(handles);

    await syncFromHydrant(env(db, bucket, vars), deps(jetstream.subscribe, repos), undefined, { startedAt });
    expect(shardCounts(db)).toEqual([3, 3, 1]);

    handles['did:plc:alice'] = 'alice.example.com';
    events = [identity(10, 'did:plc:alice', 'alice.example.com')];
    const puts = bucket.putCount;
    await syncFromHydrant(env(db, bucket, vars), deps(jetstream.subscribe, repos), undefined, {
      startedAt: new Date(startedAt.getTime() + day),
    });

    expect(repos.asked).toEqual(['did:plc:alice']);
    expect(bucket.putCount - puts).toBe(3);
    expect(await shardXml(bucket, 1)).toContain('<loc>https://tangled.org/alice.example.com/alice-1</loc>');
    expect(await shardXml(bucket, 3)).toContain('<loc>https://tangled.org/alice.example.com/alice-2</loc>');
    expect(await shardXml(bucket, 2)).not.toContain('alice');
    await expectSizesTracked(db, bucket);
  });

  it('keeps the old handle when a read fails and tries again next run', async () => {
    const db = new TestD1();
    const bucket = new MockR2Bucket();
    let events = [repo(1, 'did:plc:alice', 'tool')];
    const jetstream = claimedJetstream(() => events);
    const handles: Record<string, string> = { 'did:plc:alice': 'alice.test' };
    const repos = hydrant(handles);
    await syncFromHydrant(env(db, bucket), deps(jetstream.subscribe, repos), undefined, { startedAt });

    handles['did:plc:alice'] = 'alice.example.com';
    events = [identity(10, 'did:plc:alice', 'alice.example.com')];
    await syncFromHydrant(env(db, bucket), deps(jetstream.subscribe, down), undefined, { startedAt });
    expect(await shardXml(bucket, 1)).toContain('<loc>https://tangled.org/alice.test/tool</loc>');
    expect(handleRow(db, 'did:plc:alice')).toEqual({ handle: 'alice.test', stale: 1 });

    events = [];
    await syncFromHydrant(env(db, bucket), deps(jetstream.subscribe, repos), undefined, { startedAt });
    expect(await shardXml(bucket, 1)).toContain('<loc>https://tangled.org/alice.example.com/tool</loc>');
    expect(handleRow(db, 'did:plc:alice')).toEqual({ handle: 'alice.example.com', stale: 0 });
  });

  it('reads each owner at most once a run, even one that keeps failing', async () => {
    const db = new TestD1();
    const bucket = new MockR2Bucket();
    const owners = Object.keys(users(450));
    const jetstream = claimedJetstream(() => owners.map((did, i) => commit(i + 1, did, 'org.tangled.actor.profile', 'self')));
    const repos = hydrant();
    const asked: string[] = [];
    const broken: Repos = {
      list: (after, limit) => repos.list(after, limit),
      async get(did) {
        asked.push(did);
        if (did === owners[0]) throw new Error('bad doc');
        return null;
      },
    };

    const res = await syncFromHydrant(env(db, bucket), deps(jetstream.subscribe, broken), undefined, { startedAt });
    expect(res.lookedUpHandles).toBe(450);
    expect(asked.sort()).toEqual(owners);
    expect(handleRow(db, owners[0])).toBeUndefined();
  });

  it('stops reading after a page where nothing answered', async () => {
    const db = new TestD1();
    const bucket = new MockR2Bucket();
    const owners = Object.keys(users(250));
    const jetstream = claimedJetstream(() => owners.map((did, i) => commit(i + 1, did, 'org.tangled.actor.profile', 'self')));
    const asked: string[] = [];
    const flaky: Repos = {
      list: async () => [],
      async get(did) {
        asked.push(did);
        throw new Error('hydrant is down');
      },
    };

    const res = await syncFromHydrant(env(db, bucket), deps(jetstream.subscribe, flaky), undefined, { startedAt });
    expect(asked).toHaveLength(200);
    expect(res.lookedUpHandles).toBe(200);
    expect(db.query('SELECT did FROM handles')).toEqual([]);
    expect(bucket.store.has('sitemaps/sitemap-0001.xml.gz')).toBe(true);
  });

  it('sizes new urls with the handle they render with', async () => {
    const db = new TestD1();
    const bucket = new MockR2Bucket();
    const handle = 'a-very-long-handle-that-takes-up-a-lot-of-room.example.com';
    let events = [commit(1, 'did:plc:alice', 'org.tangled.actor.profile', 'self')];
    const jetstream = claimedJetstream(() => events);
    const repos = hydrant({ 'did:plc:alice': handle });
    const vars = { SHARD_MAX_BYTES: '700' };
    await syncFromHydrant(env(db, bucket, vars), deps(jetstream.subscribe, repos), undefined, { startedAt });

    events = [1, 2, 3, 4].map((i) => repo(10 + i, 'did:plc:alice', `repo-${i}`));
    await syncFromHydrant(env(db, bucket, vars), deps(jetstream.subscribe, repos), undefined, { startedAt });

    expect(shardCounts(db).length).toBeGreaterThan(2);
    for (const { shard } of db.query<{ shard: number }>('SELECT DISTINCT shard FROM urls')) {
      const xml = await shardXml(bucket, shard);
      expect(measureXmlBytes(xml)).toBeLessThanOrEqual(700);
    }
    expect(await shardXml(bucket, 1)).toContain(`<loc>https://tangled.org/${handle}</loc>`);
    await expectSizesTracked(db, bucket);
  });
});

describe('0002_handles migration', () => {
  it('fills in owner dids and leaves the handles to the walk', () => {
    const db = new TestD1('0002_handles.sql');
    const urls = [
      'https://tangled.org/',
      'https://tangled.org/did:plc:alice',
      'https://tangled.org/did:plc:alice/tool',
      'https://tangled.org/strings/did:plc:bob/3mtid',
      'https://tangled.org/did:web:example.com/thing',
    ];
    for (const url of urls) {
      db.sqlite.prepare("INSERT INTO urls (url, shard, lastmod, bytes) VALUES (?, 1, '2026-10-01', 100)").run(url);
    }
    db.migrate('0002_handles.sql');

    expect(db.query('SELECT url, did FROM urls ORDER BY url')).toEqual([
      { url: 'https://tangled.org/', did: null },
      { url: 'https://tangled.org/did:plc:alice', did: 'did:plc:alice' },
      { url: 'https://tangled.org/did:plc:alice/tool', did: 'did:plc:alice' },
      { url: 'https://tangled.org/did:web:example.com/thing', did: 'did:web:example.com' },
      { url: 'https://tangled.org/strings/did:plc:bob/3mtid', did: 'did:plc:bob' },
    ]);
    expect(db.query('SELECT did FROM handles')).toEqual([]);
  });
});