// The browser's `RecordStore` and `SyncStateStore`: the same interfaces the daemon implements over
// SQLite, implemented over a mirror in memory that is hydrated from — and written through to —
// IndexedDB (`idb.ts`).
//
// Two things follow from the interfaces being synchronous, and both are deliberate:
//
//   - **Hydration is the only await.** `openSpaceStores()` loads every persisted row before the
//     first `materialize()`, so the fold never sees a half-loaded space. After that a read is a map
//     lookup and a write is an array push plus a queued flush; nothing on the render path waits on a
//     transaction.
//   - **A failed write is not a failed read.** Persistence is a cache of a public record stream. If
//     IndexedDB refuses, the tab keeps working from memory and pays a cold scan next reload, which
//     is exactly the cost `phase6-ui-plan` §1 priced in.
//
// `revision` is the third piece: a counter bumped only when a `put` actually introduces something —
// a new version, an earlier revision for one already held, or a repo head that moved. The poller
// re-lists every record of a changed repo, so without it a tick that learns one new message would
// look identical to a tick that learns nothing, and every view would re-render on every poll.

import {
  collapse,
  earliest,
  select,
  type EditAnnotation,
  type RecordStore,
  type RepoSnapshot,
  type StoredRecord,
} from '@radial/core'
import type { RepoCheckpoint, SyncStateStore } from '@radial/ingest'
import { memoryDatabase, openSpaceDatabase, type Row, type SpaceDatabase, type TableName } from './idb.js'

const versionKey = (record: Pick<StoredRecord, 'did' | 'collection' | 'rkey' | 'cid'>): string =>
  `${record.did}\u0000${record.collection}\u0000${record.rkey}\u0000${record.cid}`

/** Batches writes per table and serialises the flushes, so a burst of `put`s is one transaction. */
class WriteQueue {
  readonly #pending = new Map<TableName, Map<string, Row>>()
  #chain: Promise<void> = Promise.resolve()
  #scheduled = false
  #failed = false

  constructor(readonly database: SpaceDatabase) {}

  queue(table: TableName, row: Row): void {
    if (this.#failed) return
    const rows = this.#pending.get(table) ?? new Map<string, Row>()
    rows.set(row.key, row)
    this.#pending.set(table, rows)
    if (this.#scheduled) return
    this.#scheduled = true
    queueMicrotask(() => {
      this.#scheduled = false
      this.#chain = this.#chain.then(() => this.#flush())
    })
  }

  async #flush(): Promise<void> {
    const batches = [...this.#pending]
    this.#pending.clear()
    for (const [table, rows] of batches) {
      try {
        await this.database.write(table, [...rows.values()])
      } catch {
        // Persistence is a cache; losing it costs a cold scan, not correctness. Stop queueing rather
        // than retrying a database that has already said no.
        this.#failed = true
        return
      }
    }
  }

  /** Settle every queued write. Tests await it; `close()` uses it to leave the database consistent. */
  async settle(): Promise<void> {
    // A queued flush is scheduled on a microtask, so drain until nothing new appears.
    for (let pass = 0; pass < 8; pass += 1) {
      await this.#chain
      if (this.#pending.size === 0 && !this.#scheduled) return
      if (this.#scheduled) {
        this.#scheduled = false
        this.#chain = this.#chain.then(() => this.#flush())
      }
    }
    await this.#chain
  }
}

export class PersistentRecordStore implements RecordStore {
  readonly #versions = new Map<string, StoredRecord>()
  readonly #snapshots = new Map<string, RepoSnapshot>()
  readonly #queue: WriteQueue
  #revision = 0
  #collapsedAt = -1
  #collapsed: { records: StoredRecord[]; edits: EditAnnotation[] } = { records: [], edits: [] }

  constructor(database: SpaceDatabase = memoryDatabase()) {
    this.#queue = new WriteQueue(database)
  }

  /** Bumped whenever the store learned something. A view only redraws when this moves. */
  get revision(): number {
    return this.#revision
  }

  async hydrate(database: SpaceDatabase): Promise<void> {
    for (const row of await database.load('records')) {
      const record = row.value as StoredRecord
      if (record?.uri) this.#versions.set(versionKey(record), record)
    }
    for (const row of await database.load('snapshots')) {
      const snapshot = row.value as RepoSnapshot
      if (snapshot?.did) this.#snapshots.set(snapshot.did, snapshot)
    }
    this.#revision += 1
  }

  put(record: StoredRecord): void {
    const key = versionKey(record)
    const held = this.#versions.get(key)
    let stored: StoredRecord
    if (held) {
      // The same version seen again. Only an *earlier* observed revision or an earlier first
      // sighting is news — both are observer-local stamps the fold reads, both keep the minimum,
      // and that is why a re-scan is idempotent.
      const firstSeenAt = earliest(held.firstSeenAt, record.firstSeenAt)
      const learned = record.rev < held.rev || firstSeenAt !== held.firstSeenAt
      if (!learned) return
      if (record.rev < held.rev) held.rev = record.rev
      if (firstSeenAt !== undefined) held.firstSeenAt = firstSeenAt
      stored = held
    } else {
      stored = structuredClone(record)
      this.#versions.set(key, stored)
    }
    this.#revision += 1
    this.#queue.queue('records', { key, value: stored })
  }

  #collapse(): { records: StoredRecord[]; edits: EditAnnotation[] } {
    if (this.#collapsedAt !== this.#revision) {
      this.#collapsed = collapse(structuredClone([...this.#versions.values()]))
      this.#collapsedAt = this.#revision
    }
    return this.#collapsed
  }

  records(): StoredRecord[] {
    return this.#collapse().records
  }

  edits(): EditAnnotation[] {
    return this.#collapse().edits
  }

  get(did: string, collection: string, rkey: string): StoredRecord | undefined {
    const versions = [...this.#versions.values()].filter(
      (record) => record.did === did && record.collection === collection && record.rkey === rkey,
    )
    return versions.length > 0 ? select(structuredClone(versions)).record : undefined
  }

  getSnapshot(did: string): RepoSnapshot | undefined {
    const snapshot = this.#snapshots.get(did)
    return snapshot ? structuredClone(snapshot) : undefined
  }

  setSnapshot(snapshot: RepoSnapshot): void {
    const held = this.#snapshots.get(snapshot.did)
    if (
      held &&
      held.rev === snapshot.rev &&
      held.commitCid === snapshot.commitCid &&
      held.scannerGeneration === snapshot.scannerGeneration
    ) {
      return
    }
    const stored = structuredClone(snapshot)
    this.#snapshots.set(snapshot.did, stored)
    this.#revision += 1
    this.#queue.queue('snapshots', { key: snapshot.did, value: stored })
  }

  /** Every write queued so far is durable. */
  settle(): Promise<void> {
    return this.#queue.settle()
  }

  close(): void {
    void this.#queue.settle()
  }
}

export class PersistentSyncStateStore implements SyncStateStore {
  readonly #checkpoints = new Map<string, RepoCheckpoint>()
  readonly #queue: WriteQueue

  constructor(database: SpaceDatabase = memoryDatabase()) {
    this.#queue = new WriteQueue(database)
  }

  async hydrate(database: SpaceDatabase): Promise<void> {
    for (const row of await database.load('checkpoints')) {
      const checkpoint = row.value as RepoCheckpoint
      if (checkpoint?.did) this.#checkpoints.set(checkpoint.did, checkpoint)
    }
  }

  get(did: string): RepoCheckpoint | undefined {
    const checkpoint = this.#checkpoints.get(did)
    return checkpoint ? structuredClone(checkpoint) : undefined
  }

  set(checkpoint: RepoCheckpoint): void {
    const stored = structuredClone(checkpoint)
    this.#checkpoints.set(checkpoint.did, stored)
    this.#queue.queue('checkpoints', { key: checkpoint.did, value: stored })
  }

  all(): RepoCheckpoint[] {
    return [...this.#checkpoints.values()].map((checkpoint) => structuredClone(checkpoint))
  }

  settle(): Promise<void> {
    return this.#queue.settle()
  }

  close(): void {
    void this.#queue.settle()
  }
}

export interface SpaceStores {
  records: PersistentRecordStore
  state: PersistentSyncStateStore
  database: SpaceDatabase
  close(): void
}

/**
 * Open a space's stores and fill them from disk before anything folds. This is the whole of the
 * "warm reload" claim: with the record versions AND the repo snapshots AND the sync checkpoints all
 * back in place, `RepoPoller` short-circuits on an unchanged `getLatestCommit` and never lists a
 * collection.
 */
export async function openSpaceStores(
  spaceUri: string,
  open: (uri: string) => Promise<SpaceDatabase> = openSpaceDatabase,
): Promise<SpaceStores> {
  const database = await open(spaceUri)
  const records = new PersistentRecordStore(database)
  const state = new PersistentSyncStateStore(database)
  await records.hydrate(database)
  await state.hydrate(database)
  return {
    records,
    state,
    database,
    close() {
      records.close()
      state.close()
      void Promise.all([records.settle(), state.settle()]).then(() => database.close())
    },
  }
}