// Cloudflare provides this runtime module inside Workers. The root Node test
// program intentionally avoids Worker globals because they replace standard
// web types like `Response.json`.
// @ts-ignore
import { DurableObject as CloudflareDurableObject } from 'cloudflare:workers'

import * as Db from './db/Db.js'
import * as Funding from './internal/funding/index.js'
import * as Metrics from './Metrics.js'
import * as Store from './internal/Store.js'

type State = ConstructorParameters<typeof CloudflareDurableObject>[0]
type Environment = ConstructorParameters<typeof CloudflareDurableObject>[1]

/** Durable Object class that exposes a Tempo API store over Cloudflare RPC. */
export class DurableObject extends CloudflareDurableObject {
  #store: Store.Store

  constructor(ctx: State, env: Environment) {
    // The package also typechecks without Worker globals, so `Environment`
    // resolves through constructor parameters instead of the generated Env type.
    super(ctx, env as never)
    const { storage } = ctx as { storage: Store.durableObject.Storage }
    this.#store = Store.durableObject(storage)
  }

  /** Deletes a key. */
  async delete(key: string) {
    await this.#store.delete(key)
  }

  /** Deletes a key only when its value matches `expected`. */
  async deleteIf(key: string, expected: null | string) {
    return this.#store.deleteIf!(key, expected)
  }

  /** Gets a string value. Returns `null` for missing or expired keys. */
  async get(key: string) {
    return this.#store.get(key)
  }

  /**
   * Increments an integer counter and returns the new value.
   *
   * The helper's get + put fallback runs inside the object, so the caller
   * pays one RPC round trip instead of a serial get + put, and the counter
   * cannot race: the object is single-threaded and its input gate holds
   * other events while the storage operations are in flight.
   */
  async increment(key: string, options?: Store.Store.PutOptions) {
    return Store.increment(this.#store, key, options)
  }

  /** Lists keys. */
  async list(options?: Parameters<Store.Store['list']>[0]) {
    return this.#store.list(options)
  }

  /** Writes a string value, optionally with a time-to-live. */
  async put(key: string, value: string, options?: Store.Store.PutOptions) {
    await this.#store.put(key, value, options)
  }

  /**
   * Atomically compares-and-swaps a value.
   *
   * The compare and write run on the object's single thread behind its input
   * gate, so concurrent callers cannot interleave between them.
   */
  async swap(key: string, expected: null | string, next: string, options?: Store.Store.PutOptions) {
    return this.#store.swap!(key, expected, next, options)
  }
}

const fundingSourceAlarmIntervalMs = 500
const fundingSourceRetryIntervalMs = 2_000

type FundingSourceConfiguration = {
  chainId: string
  rpcUrls: readonly string[]
  tokenAddress: string
}

type FundingSourceRoute = Funding.SourceObservation.Route

type FundingSourceRegisterOptions = FundingSourceConfiguration & {
  route: FundingSourceRoute
}

type FundingSourceSyncOptions = FundingSourceConfiguration & {
  routes: readonly FundingSourceRoute[]
}

type FundingSourceConfigurationRow = {
  value: string
}

type FundingSourceRouteRow = {
  address: string
  id: string
}

type FundingSourceStateRow = {
  value: string
}

/** Cloudflare Durable Object that detects EVM funding deposits before provider indexing. */
export class FundingSourceObserver<
  environment extends FundingSourceObserver.Environment = FundingSourceObserver.Environment,
> extends CloudflareDurableObject {
  /** Durable Object storage and alarm context. */
  protected context: State
  /** Cloudflare bindings used by the observer. */
  protected environment: environment

  constructor(ctx: State, env: environment) {
    super(ctx, env as never)
    this.context = ctx
    this.environment = env
    void ctx.blockConcurrencyWhile(async () => {
      ctx.storage.sql.exec(
        'CREATE TABLE IF NOT EXISTS configuration (id INTEGER PRIMARY KEY CHECK (id = 1), value TEXT NOT NULL)',
      )
      ctx.storage.sql.exec('CREATE TABLE IF NOT EXISTS backfills (id TEXT PRIMARY KEY)')
      ctx.storage.sql.exec(
        'CREATE TABLE IF NOT EXISTS routes (id TEXT PRIMARY KEY, address TEXT NOT NULL UNIQUE)',
      )
      ctx.storage.sql.exec(
        'CREATE TABLE IF NOT EXISTS state (key TEXT PRIMARY KEY, value TEXT NOT NULL)',
      )
    })
  }

  /** Registers one newly created deposit address and starts source observation. */
  async register(options: FundingSourceRegisterOptions): Promise<void> {
    const configuration = serializeFundingSourceConfiguration(options)
    const route = this.context.storage.sql
      .exec<FundingSourceRouteRow>('SELECT id, address FROM routes WHERE id = ?', options.route.id)
      .toArray()[0]
    this.context.storage.transactionSync(() => {
      this.context.storage.sql.exec(
        'INSERT OR REPLACE INTO configuration (id, value) VALUES (1, ?)',
        configuration,
      )
      this.context.storage.sql.exec(
        'INSERT OR REPLACE INTO routes (id, address) VALUES (?, ?)',
        options.route.id,
        options.route.address.toLowerCase(),
      )
      if (!route || route.address.toLowerCase() !== options.route.address.toLowerCase())
        this.context.storage.sql.exec(
          'INSERT OR IGNORE INTO backfills (id) VALUES (?)',
          options.route.id,
        )
    })
    await this.#ensureAlarm()
  }

  /** Replaces the active routes from the database and repairs alarm scheduling. */
  async sync(options: FundingSourceSyncOptions): Promise<void> {
    const configuration = serializeFundingSourceConfiguration(options)
    const existing = new Map(
      this.context.storage.sql
        .exec<FundingSourceRouteRow>('SELECT id, address FROM routes ORDER BY id')
        .toArray()
        .map((route: FundingSourceRouteRow) => [route.id, route.address.toLowerCase()] as const),
    )
    const changed = options.routes.filter(
      (route) => existing.get(route.id) !== route.address.toLowerCase(),
    )
    this.context.storage.transactionSync(() => {
      this.context.storage.sql.exec(
        'INSERT OR REPLACE INTO configuration (id, value) VALUES (1, ?)',
        configuration,
      )
      this.context.storage.sql.exec('DELETE FROM routes')
      for (const route of options.routes)
        this.context.storage.sql.exec(
          'INSERT INTO routes (id, address) VALUES (?, ?)',
          route.id,
          route.address.toLowerCase(),
        )
      this.context.storage.sql.exec('DELETE FROM backfills WHERE id NOT IN (SELECT id FROM routes)')
      for (const route of changed)
        this.context.storage.sql.exec('INSERT OR IGNORE INTO backfills (id) VALUES (?)', route.id)
    })
    if (options.routes.length === 0) {
      await this.context.storage.deleteAlarm()
      return
    }
    await this.#ensureAlarm()
  }

  /** Scans confirmed source blocks and persists newly observed deposits. */
  async alarm(): Promise<void> {
    const metrics = this.createMetrics()
    const configuration = this.#readConfiguration()
    const routes: FundingSourceRouteRow[] = this.context.storage.sql
      .exec<FundingSourceRouteRow>('SELECT id, address FROM routes ORDER BY id')
      .toArray()
    if (!configuration || routes.length === 0) return

    try {
      const cursor = this.context.storage.sql
        .exec<FundingSourceStateRow>("SELECT value FROM state WHERE key = 'cursor'")
        .toArray()[0]
      const result = await Funding.SourceObservation.scan({
        chainId: configuration.chainId,
        ...(cursor ? { cursor: BigInt(cursor.value) } : {}),
        routes,
        rpcUrls: configuration.rpcUrls,
        tokenAddress: configuration.tokenAddress,
      })
      const backfills = this.context.storage.sql
        .exec<FundingSourceRouteRow>(
          'SELECT routes.id, routes.address FROM routes INNER JOIN backfills ON backfills.id = routes.id ORDER BY routes.id',
        )
        .toArray()
      // Backfill new routes independently so the main cursor retains any older shard backlog.
      const backfill = await (async () => {
        if (!cursor || backfills.length === 0) return undefined
        return Funding.SourceObservation.scan({
          chainId: configuration.chainId,
          routes: backfills,
          rpcUrls: configuration.rpcUrls,
          tokenAddress: configuration.tokenAddress,
        })
      })()
      const database = Db.get(() =>
        Db.postgres({ connectionString: this.environment.HYPERDRIVE.connectionString }),
      )
      const routesById = new Map(routes.map((route) => [route.id, route] as const))
      const observations = new Map(
        [...result.observations, ...(backfill?.observations ?? [])].map((observation) => [
          `${observation.addressId}:${observation.transactionHash.toLowerCase()}:${observation.transferIndex}`,
          observation,
        ]),
      ).values()
      for (const observation of observations) {
        const route = routesById.get(observation.addressId)
        if (!route) continue
        const persisted = await Funding.SourceObservation.observe(database, {
          ...observation,
          recipient: route.address,
        })
        if (
          persisted.type === 'ignored' ||
          persisted.record.providerRequestId !== null ||
          Funding.Deposit.isTerminal(persisted.record.status)
        )
          continue
        await this.environment.FUNDING_RECONCILIATION_QUEUE.send({
          addressId: observation.addressId,
          trigger: 'chain',
          type: 'funding:deposit-address:reconcile',
        })
        metrics.count('funding_source_observer_deposit_count', 1, {
          chain: configuration.chainId,
          outcome: persisted.type,
        })
      }
      this.context.storage.transactionSync(() => {
        this.context.storage.sql.exec(
          "INSERT OR REPLACE INTO state (key, value) VALUES ('cursor', ?)",
          result.cursor.toString(),
        )
        for (const route of backfills)
          this.context.storage.sql.exec('DELETE FROM backfills WHERE id = ?', route.id)
      })
      metrics.count('funding_source_observer_scan_count', 1, {
        chain: configuration.chainId,
        outcome: 'completed',
      })
      metrics.gauge(
        'funding_source_observer_block_lag',
        Math.max(0, Number(result.head - result.cursor)),
        { chain: configuration.chainId },
      )
      await this.context.storage.setAlarm(
        result.cursor < result.head ? Date.now() : Date.now() + fundingSourceAlarmIntervalMs,
      )
    } catch (cause) {
      metrics.count('funding_source_observer_scan_count', 1, {
        chain: configuration.chainId,
        outcome: 'failed',
      })
      this.capture(cause)
      await this.context.storage.setAlarm(Date.now() + fundingSourceRetryIntervalMs)
    } finally {
      metrics.flush()
    }
  }

  /** Reports one source-observer failure. */
  protected capture(cause: unknown): void {
    console.error(cause)
  }

  /** Creates the operational metric sink used by one alarm invocation. */
  protected createMetrics(): Metrics.Metrics {
    return Metrics.cloudflare({
      enabled: Boolean(this.environment.ENVIRONMENT),
      environment: this.environment.ENVIRONMENT ?? 'development',
      service: 'tapimo',
    })
  }

  #configuration(): string | undefined {
    return this.context.storage.sql
      .exec<FundingSourceConfigurationRow>('SELECT value FROM configuration WHERE id = 1')
      .toArray()[0]?.value
  }

  async #ensureAlarm(): Promise<void> {
    if ((await this.context.storage.getAlarm()) === null)
      await this.context.storage.setAlarm(Date.now())
  }

  #readConfiguration(): FundingSourceConfiguration | undefined {
    const value = this.#configuration()
    return value ? (JSON.parse(value) as FundingSourceConfiguration) : undefined
  }
}

export declare namespace FundingSourceObserver {
  /** Cloudflare bindings and deployment metadata used by the observer. */
  type Environment = {
    /** Deployment environment included in operational metrics. */
    ENVIRONMENT?: string | undefined
    /** Queue that receives deposit-address reconciliation work. */
    FUNDING_RECONCILIATION_QUEUE: {
      /** Enqueues one chain-detected address for reconciliation. */
      send(message: Funding.Provider.Webhook.receive.Dispatchable): Promise<unknown>
    }
    /** Authoritative Postgres connection exposed through Hyperdrive. */
    HYPERDRIVE: {
      /** PostgreSQL connection string supplied by the Hyperdrive binding. */
      connectionString: string
    }
  }
}

function serializeFundingSourceConfiguration(options: FundingSourceConfiguration): string {
  return JSON.stringify({
    chainId: options.chainId,
    rpcUrls: options.rpcUrls,
    tokenAddress: options.tokenAddress.toLowerCase(),
  } satisfies FundingSourceConfiguration)
}