|
| 1 | +import { RevisionConflict, type MarkdownPointer, type RevisionCoordinator, type RevisionLease, type RevisionState } from "./revision-coordinator.js" |
| 2 | + |
| 3 | +interface StateRow { |
| 4 | + revision: string | null |
| 5 | + manifest_key: string | null |
| 6 | + persisted_at: string | null |
| 7 | + version: number |
| 8 | + lease_token: string | null |
| 9 | + lease_base_revision: string | null |
| 10 | + lease_version: number | null |
| 11 | + lease_expires_at: number | null |
| 12 | +} |
| 13 | + |
| 14 | +const SITE_ID = "default" |
| 15 | +const LEASE_MS = 90_000 |
| 16 | +const schemaReady = new WeakMap<object, Promise<void>>() |
| 17 | + |
| 18 | +export class D1RevisionCoordinator implements RevisionCoordinator { |
| 19 | + constructor(private readonly database: D1Database, private readonly leaseMs = LEASE_MS) {} |
| 20 | + |
| 21 | + state(): Promise<RevisionState> { |
| 22 | + return readWordPressState(this.database) |
| 23 | + } |
| 24 | + |
| 25 | + acquire(): Promise<RevisionLease> { |
| 26 | + return beginStateLease(this.database, this.leaseMs) |
| 27 | + } |
| 28 | + |
| 29 | + async release(lease: RevisionLease): Promise<void> { |
| 30 | + await releaseStateLease(this.database, lease) |
| 31 | + } |
| 32 | + |
| 33 | + async abort(lease: RevisionLease): Promise<void> { |
| 34 | + await abortStateLease(this.database, lease) |
| 35 | + } |
| 36 | + |
| 37 | + commit(lease: RevisionLease, pointer: MarkdownPointer): Promise<{ pointer: MarkdownPointer; version: number }> { |
| 38 | + return commitStateLease(this.database, lease, pointer) |
| 39 | + } |
| 40 | + |
| 41 | + async reset(): Promise<void> { |
| 42 | + await resetWordPressState(this.database) |
| 43 | + } |
| 44 | +} |
| 45 | + |
| 46 | +async function readWordPressState(database: D1Database): Promise<RevisionState> { |
| 47 | + await ensureSchema(database) |
| 48 | + const row = await readRow(database) |
| 49 | + return { |
| 50 | + schema: "wp-codebox/cloudflare-wordpress-state/v2", |
| 51 | + store: "d1", |
| 52 | + pointer: pointerFromRow(row), |
| 53 | + version: row.version, |
| 54 | + } |
| 55 | +} |
| 56 | + |
| 57 | +async function beginStateLease(database: D1Database, leaseMs = LEASE_MS): Promise<RevisionLease> { |
| 58 | + await ensureSchema(database) |
| 59 | + const now = Date.now() |
| 60 | + const token = crypto.randomUUID() |
| 61 | + const expiresAt = now + leaseMs |
| 62 | + const result = await database.prepare(`UPDATE wp_codebox_state |
| 63 | + SET lease_token = ?, lease_base_revision = revision, lease_version = version, lease_expires_at = ? |
| 64 | + WHERE site_id = ? AND (lease_token IS NULL OR lease_expires_at <= ?)`) |
| 65 | + .bind(token, expiresAt, SITE_ID, now).run() |
| 66 | + if (result.meta.changes !== 1) { |
| 67 | + const active = await readRow(database) |
| 68 | + throw new RevisionConflict("A canonical WordPress lease is active.", active.lease_expires_at ?? undefined) |
| 69 | + } |
| 70 | + const row = await readRow(database) |
| 71 | + if (row.lease_token !== token || row.lease_version === null) throw new RevisionConflict("The canonical WordPress lease was not acquired.") |
| 72 | + return { token, pointer: pointerFromRow(row), version: row.lease_version, expiresAt } |
| 73 | +} |
| 74 | + |
| 75 | +async function releaseStateLease(database: D1Database, lease: RevisionLease): Promise<{ released: true }> { |
| 76 | + await finishLease(database, lease, "release") |
| 77 | + return { released: true } |
| 78 | +} |
| 79 | + |
| 80 | +async function abortStateLease(database: D1Database, lease: RevisionLease): Promise<{ aborted: true }> { |
| 81 | + await finishLease(database, lease, "abort") |
| 82 | + return { aborted: true } |
| 83 | +} |
| 84 | + |
| 85 | +async function commitStateLease(database: D1Database, lease: RevisionLease, pointer: MarkdownPointer): Promise<{ pointer: MarkdownPointer; version: number }> { |
| 86 | + validatePointer(pointer) |
| 87 | + await ensureSchema(database) |
| 88 | + const baseRevision = lease.pointer?.revision ?? null |
| 89 | + const result = await database.prepare(`UPDATE wp_codebox_state |
| 90 | + SET revision = ?, manifest_key = ?, persisted_at = ?, version = version + 1, |
| 91 | + lease_token = NULL, lease_base_revision = NULL, lease_version = NULL, lease_expires_at = NULL |
| 92 | + WHERE site_id = ? AND lease_token = ? AND lease_expires_at > ? AND version = ? AND lease_version = ? |
| 93 | + AND ((revision IS NULL AND ? IS NULL) OR revision = ?) |
| 94 | + AND ((lease_base_revision IS NULL AND ? IS NULL) OR lease_base_revision = ?)`) |
| 95 | + .bind(pointer.revision, pointer.manifestKey, pointer.persistedAt, SITE_ID, lease.token, Date.now(), lease.version, lease.version, |
| 96 | + baseRevision, baseRevision, baseRevision, baseRevision).run() |
| 97 | + if (result.meta.changes !== 1) throw new RevisionConflict("The canonical pointer changed before D1 promotion.") |
| 98 | + return { pointer, version: lease.version + 1 } |
| 99 | +} |
| 100 | + |
| 101 | +async function resetWordPressState(database: D1Database): Promise<{ reset: true }> { |
| 102 | + await ensureSchema(database) |
| 103 | + await database.prepare(`UPDATE wp_codebox_state |
| 104 | + SET revision = NULL, manifest_key = NULL, persisted_at = NULL, version = version + 1, |
| 105 | + lease_token = NULL, lease_base_revision = NULL, lease_version = NULL, lease_expires_at = NULL |
| 106 | + WHERE site_id = ?`).bind(SITE_ID).run() |
| 107 | + return { reset: true } |
| 108 | +} |
| 109 | + |
| 110 | +async function finishLease(database: D1Database, lease: RevisionLease, action: "release" | "abort"): Promise<void> { |
| 111 | + await ensureSchema(database) |
| 112 | + const result = await database.prepare(`UPDATE wp_codebox_state |
| 113 | + SET lease_token = NULL, lease_base_revision = NULL, lease_version = NULL, lease_expires_at = NULL |
| 114 | + WHERE site_id = ? AND lease_token = ? AND lease_expires_at > ?`) |
| 115 | + .bind(SITE_ID, lease.token, Date.now()).run() |
| 116 | + if (result.meta.changes !== 1) throw new RevisionConflict(`The canonical WordPress lease cannot ${action} because it expired or changed.`) |
| 117 | +} |
| 118 | + |
| 119 | +async function readRow(database: D1Database): Promise<StateRow> { |
| 120 | + const row = await database.prepare(`SELECT revision, manifest_key, persisted_at, version, |
| 121 | + lease_token, lease_base_revision, lease_version, lease_expires_at |
| 122 | + FROM wp_codebox_state WHERE site_id = ?`).bind(SITE_ID).first<StateRow>() |
| 123 | + if (!row) throw new Error("D1 WordPress state row is unavailable.") |
| 124 | + return row |
| 125 | +} |
| 126 | + |
| 127 | +function pointerFromRow(row: StateRow): MarkdownPointer | null { |
| 128 | + if (row.revision === null && row.manifest_key === null && row.persisted_at === null) return null |
| 129 | + const pointer = { revision: row.revision, manifestKey: row.manifest_key, persistedAt: row.persisted_at } |
| 130 | + validatePointer(pointer) |
| 131 | + return pointer as MarkdownPointer |
| 132 | +} |
| 133 | + |
| 134 | +function validatePointer(pointer: unknown): asserts pointer is MarkdownPointer { |
| 135 | + if (!pointer || typeof pointer !== "object") throw new RevisionConflict("A complete canonical pointer is required for D1 promotion.") |
| 136 | + const candidate = pointer as Partial<MarkdownPointer> |
| 137 | + if (typeof candidate.revision !== "string" || typeof candidate.manifestKey !== "string" || typeof candidate.persistedAt !== "string") { |
| 138 | + throw new RevisionConflict("A complete canonical pointer is required for D1 promotion.") |
| 139 | + } |
| 140 | +} |
| 141 | + |
| 142 | +function ensureSchema(database: D1Database): Promise<void> { |
| 143 | + const key = database as object |
| 144 | + const existing = schemaReady.get(key) |
| 145 | + if (existing) return existing |
| 146 | + const pending = (async () => { |
| 147 | + await database.prepare(`CREATE TABLE IF NOT EXISTS wp_codebox_state ( |
| 148 | + site_id TEXT PRIMARY KEY, |
| 149 | + revision TEXT, |
| 150 | + manifest_key TEXT, |
| 151 | + persisted_at TEXT, |
| 152 | + version INTEGER NOT NULL DEFAULT 0, |
| 153 | + lease_token TEXT, |
| 154 | + lease_base_revision TEXT, |
| 155 | + lease_version INTEGER, |
| 156 | + lease_expires_at INTEGER |
| 157 | + )`).run() |
| 158 | + await database.prepare(`INSERT OR IGNORE INTO wp_codebox_state (site_id, version) VALUES (?, 0)`).bind(SITE_ID).run() |
| 159 | + })() |
| 160 | + schemaReady.set(key, pending) |
| 161 | + pending.catch(() => schemaReady.delete(key)) |
| 162 | + return pending |
| 163 | +} |
0 commit comments