|
| 1 | +import type { SiteContext } from "./site-context.js" |
| 2 | + |
| 3 | +export type AllocationLifecycleState = "active" | "deleting" | "tombstoned" |
| 4 | +export interface AllocationIdentity { siteId: string; generation: number } |
| 5 | +export interface AllocationRetentionPolicy { ttlMs: number; retainMs: number } |
| 6 | +export interface AllocationDeletionReceipt { schema: "wp-codebox/cloudflare-allocation-deletion-receipt/v1"; siteId: string; generation: number; deletedObjects: number; completedAt: string } |
| 7 | +export interface AllocationLifecycle { identity: AllocationIdentity; principal: string; state: AllocationLifecycleState; expiresAt: number; retainUntil: number; mutationFence: number; operationFence: number; cleanupCursor: string | null; deletedObjects: number; receipt: AllocationDeletionReceipt | null } |
| 8 | + |
| 9 | +const RECEIPT_SCHEMA = "wp-codebox/cloudflare-allocation-deletion-receipt/v1" as const |
| 10 | +const schemaReady = new WeakMap<object, Promise<void>>() |
| 11 | + |
| 12 | +/** Durable lifecycle policy for elastic Cloudflare site allocations. */ |
| 13 | +export class CloudflareAllocationLifecycle { |
| 14 | + constructor(private readonly database: D1Database, private readonly policy: AllocationRetentionPolicy = { ttlMs: 24 * 60 * 60 * 1000, retainMs: 60 * 60 * 1000 }) { |
| 15 | + if (!Number.isSafeInteger(policy.ttlMs) || !Number.isSafeInteger(policy.retainMs) || policy.ttlMs < 1 || policy.retainMs < 0) throw new Error("Allocation lifecycle policy is invalid.") |
| 16 | + } |
| 17 | + |
| 18 | + async initialize(): Promise<void> { await ensureSchema(this.database) } |
| 19 | + async create(site: SiteContext, principal: string, now = Date.now()): Promise<AllocationLifecycle> { |
| 20 | + await this.initialize() |
| 21 | + const identity = allocationIdentity(site.id) |
| 22 | + await this.database.prepare("INSERT OR IGNORE INTO wp_codebox_site_lifecycles (site_id, generation, principal, state, expires_at, retain_until, mutation_fence, operation_fence, cleanup_cursor, deleted_objects, created_at, updated_at) VALUES (?, ?, ?, 'active', ?, ?, 1, 1, NULL, 0, ?, ?)").bind(site.id, identity.generation, principal, now + this.policy.ttlMs, now + this.policy.ttlMs + this.policy.retainMs, now, now).run() |
| 23 | + const lifecycle = await this.get(identity) |
| 24 | + if (!lifecycle || lifecycle.principal !== principal) throw new AllocationLifecycleConflict("Allocation identity conflicts with an existing owner.") |
| 25 | + return lifecycle |
| 26 | + } |
| 27 | + async get(identity: AllocationIdentity): Promise<AllocationLifecycle | null> { |
| 28 | + await this.initialize() |
| 29 | + const row = await this.database.prepare("SELECT site_id, generation, principal, state, expires_at, retain_until, mutation_fence, operation_fence, cleanup_cursor, deleted_objects, receipt_json FROM wp_codebox_site_lifecycles WHERE site_id = ? AND generation = ?").bind(identity.siteId, identity.generation).first<LifecycleRow>() |
| 30 | + return row ? hydrate(row) : null |
| 31 | + } |
| 32 | + async renew(identity: AllocationIdentity, principal: string, now = Date.now()): Promise<AllocationLifecycle> { |
| 33 | + await this.initialize() |
| 34 | + const result = await this.database.prepare("UPDATE wp_codebox_site_lifecycles SET expires_at = ?, retain_until = ?, updated_at = ? WHERE site_id = ? AND generation = ? AND principal = ? AND state = 'active' AND expires_at > ?").bind(now + this.policy.ttlMs, now + this.policy.ttlMs + this.policy.retainMs, now, identity.siteId, identity.generation, principal, now).run() |
| 35 | + if (result.meta.changes !== 1) throw new AllocationLifecycleConflict("Allocation is not renewable by this owner.") |
| 36 | + return (await this.get(identity))! |
| 37 | + } |
| 38 | + async beginMutation(identity: AllocationIdentity, principal: string, now = Date.now()): Promise<number> { |
| 39 | + await this.initialize() |
| 40 | + const result = await this.database.prepare("UPDATE wp_codebox_site_lifecycles SET mutation_fence = mutation_fence + 1, updated_at = ? WHERE site_id = ? AND generation = ? AND principal = ? AND state = 'active' AND expires_at > ?").bind(now, identity.siteId, identity.generation, principal, now).run() |
| 41 | + if (result.meta.changes !== 1) throw new AllocationLifecycleConflict("Allocation is not mutable by this owner.") |
| 42 | + return (await this.get(identity))!.mutationFence |
| 43 | + } |
| 44 | + async assertMutation(identity: AllocationIdentity, fence: number, now = Date.now()): Promise<void> { |
| 45 | + const lifecycle = await this.get(identity) |
| 46 | + if (!lifecycle || lifecycle.state !== "active" || lifecycle.expiresAt <= now || lifecycle.mutationFence !== fence) throw new AllocationLifecycleConflict("Allocation mutation fence changed.") |
| 47 | + } |
| 48 | + async beginDeletion(identity: AllocationIdentity, principal: string | null, now = Date.now()): Promise<number> { |
| 49 | + await this.initialize() |
| 50 | + const owner = principal === null ? "" : " AND principal = ?" |
| 51 | + const values: unknown[] = [now, identity.siteId, identity.generation] |
| 52 | + if (principal !== null) values.push(principal) |
| 53 | + const result = await this.database.prepare(`UPDATE wp_codebox_site_lifecycles SET state = 'deleting', mutation_fence = mutation_fence + 1, operation_fence = operation_fence + 1, cleanup_cursor = NULL, updated_at = ? WHERE site_id = ? AND generation = ?${owner} AND state = 'active'`).bind(...values).run() |
| 54 | + if (result.meta.changes !== 1) throw new AllocationLifecycleConflict("Allocation is not deletable by this owner.") |
| 55 | + return (await this.get(identity))!.operationFence |
| 56 | + } |
| 57 | + async expire(now = Date.now(), limit = 8): Promise<AllocationIdentity[]> { |
| 58 | + await this.initialize() |
| 59 | + const rows = await this.database.prepare("SELECT site_id, generation FROM wp_codebox_site_lifecycles WHERE state = 'active' AND expires_at <= ? ORDER BY expires_at LIMIT ?").bind(now, limit).all<Pick<LifecycleRow, "site_id" | "generation">>() |
| 60 | + const expired: AllocationIdentity[] = [] |
| 61 | + for (const row of rows.results) { |
| 62 | + try { await this.beginDeletion({ siteId: row.site_id, generation: row.generation }, null, now); expired.push({ siteId: row.site_id, generation: row.generation }) } catch { /* A competing owner or sweeper already fenced it. */ } |
| 63 | + } |
| 64 | + return expired |
| 65 | + } |
| 66 | + async pendingDeletions(limit = 8): Promise<Array<AllocationLifecycle>> { |
| 67 | + await this.initialize() |
| 68 | + const rows = await this.database.prepare("SELECT site_id, generation, principal, state, expires_at, retain_until, mutation_fence, operation_fence, cleanup_cursor, deleted_objects, receipt_json FROM wp_codebox_site_lifecycles WHERE state = 'deleting' ORDER BY updated_at LIMIT ?").bind(limit).all<LifecycleRow>() |
| 69 | + return rows.results.map(hydrate) |
| 70 | + } |
| 71 | + async reclaim(bucket: Pick<R2Bucket, "list" | "delete">, identity: AllocationIdentity, operationFence: number, limit = 100, now = Date.now()): Promise<AllocationLifecycle> { |
| 72 | + if (!Number.isInteger(limit) || limit < 1 || limit > 1000) throw new Error("Allocation reclamation limit is invalid.") |
| 73 | + const lifecycle = await this.get(identity) |
| 74 | + if (!lifecycle || lifecycle.state !== "deleting" || lifecycle.operationFence !== operationFence) throw new AllocationLifecycleConflict("Allocation deletion fence changed.") |
| 75 | + const page = await bucket.list({ prefix: `sites/${identity.siteId}/`, cursor: lifecycle.cleanupCursor ?? undefined, limit }) |
| 76 | + if (page.objects.length) await bucket.delete(page.objects.map((object) => object.key)) |
| 77 | + const cursor = page.truncated ? page.cursor : null |
| 78 | + const updated = await this.database.prepare("UPDATE wp_codebox_site_lifecycles SET cleanup_cursor = ?, deleted_objects = deleted_objects + ?, updated_at = ? WHERE site_id = ? AND generation = ? AND state = 'deleting' AND operation_fence = ?").bind(cursor, page.objects.length, now, identity.siteId, identity.generation, operationFence).run() |
| 79 | + if (updated.meta.changes !== 1) throw new AllocationLifecycleConflict("Allocation deletion fence changed.") |
| 80 | + if (cursor) return (await this.get(identity))! |
| 81 | + const receipt: AllocationDeletionReceipt = { schema: RECEIPT_SCHEMA, siteId: identity.siteId, generation: identity.generation, deletedObjects: lifecycle.deletedObjects + page.objects.length, completedAt: new Date(now).toISOString() } |
| 82 | + await this.database.batch([ |
| 83 | + this.database.prepare("INSERT OR IGNORE INTO wp_codebox_site_deletion_receipts (site_id, generation, receipt_json, created_at) VALUES (?, ?, ?, ?)").bind(identity.siteId, identity.generation, JSON.stringify(receipt), now), |
| 84 | + this.database.prepare("UPDATE wp_codebox_site_lifecycles SET state = 'tombstoned', cleanup_cursor = NULL, receipt_json = ?, terminal_at = ?, updated_at = ? WHERE site_id = ? AND generation = ? AND state = 'deleting' AND operation_fence = ?").bind(JSON.stringify(receipt), now, now, identity.siteId, identity.generation, operationFence), |
| 85 | + ]) |
| 86 | + return (await this.get(identity))! |
| 87 | + } |
| 88 | +} |
| 89 | + |
| 90 | +export class AllocationLifecycleConflict extends Error {} |
| 91 | +export function allocationIdentity(siteId: string): AllocationIdentity { |
| 92 | + const match = /-g([1-9][0-9]*)-[a-f0-9]{16}$/.exec(siteId) |
| 93 | + return { siteId, generation: match ? Number(match[1]) : 1 } |
| 94 | +} |
| 95 | + |
| 96 | +interface LifecycleRow { site_id: string; generation: number; principal: string; state: AllocationLifecycleState; expires_at: number; retain_until: number; mutation_fence: number; operation_fence: number; cleanup_cursor: string | null; deleted_objects: number; receipt_json: string | null } |
| 97 | +function hydrate(row: LifecycleRow): AllocationLifecycle { return { identity: { siteId: row.site_id, generation: row.generation }, principal: row.principal, state: row.state, expiresAt: row.expires_at, retainUntil: row.retain_until, mutationFence: row.mutation_fence, operationFence: row.operation_fence, cleanupCursor: row.cleanup_cursor, deletedObjects: row.deleted_objects, receipt: row.receipt_json ? JSON.parse(row.receipt_json) as AllocationDeletionReceipt : null } } |
| 98 | +async function ensureSchema(database: D1Database): Promise<void> { |
| 99 | + let pending = schemaReady.get(database as object) |
| 100 | + if (!pending) { |
| 101 | + pending = (async () => { |
| 102 | + await database.prepare("CREATE TABLE IF NOT EXISTS wp_codebox_site_lifecycles (site_id TEXT NOT NULL, generation INTEGER NOT NULL, principal TEXT NOT NULL, state TEXT NOT NULL CHECK (state IN ('active','deleting','tombstoned')), expires_at INTEGER NOT NULL, retain_until INTEGER NOT NULL, mutation_fence INTEGER NOT NULL, operation_fence INTEGER NOT NULL, cleanup_cursor TEXT, deleted_objects INTEGER NOT NULL, receipt_json TEXT, terminal_at INTEGER, created_at INTEGER NOT NULL, updated_at INTEGER NOT NULL, PRIMARY KEY (site_id, generation))").run() |
| 103 | + await database.prepare("CREATE TABLE IF NOT EXISTS wp_codebox_site_deletion_receipts (site_id TEXT NOT NULL, generation INTEGER NOT NULL, receipt_json TEXT NOT NULL, created_at INTEGER NOT NULL, PRIMARY KEY (site_id, generation))").run() |
| 104 | + await database.prepare("CREATE INDEX IF NOT EXISTS wp_codebox_site_lifecycle_expiry ON wp_codebox_site_lifecycles(state, expires_at)").run() |
| 105 | + })() |
| 106 | + schemaReady.set(database as object, pending) |
| 107 | + pending.catch(() => schemaReady.delete(database as object)) |
| 108 | + } |
| 109 | + await pending |
| 110 | +} |
0 commit comments