Move shared cycle state to Neon fallback

This commit is contained in:
Theodore Song
2026-08-02 14:45:06 -04:00
parent ecbdbf4cbf
commit 6233efc098
2 changed files with 87 additions and 9 deletions
+40
View File
@@ -2,6 +2,7 @@ import { neon } from "@neondatabase/serverless";
let sqlClient; let sqlClient;
let schemaReady; let schemaReady;
let sharedStateSchemaReady;
export function databaseUrl() { export function databaseUrl() {
return process.env.DATABASE_URL || process.env.NEON_DATABASE_URL || ""; return process.env.DATABASE_URL || process.env.NEON_DATABASE_URL || "";
@@ -20,6 +21,45 @@ export function sql() {
return sqlClient; return sqlClient;
} }
async function ensureSharedStateSchema() {
if (sharedStateSchemaReady) return;
const db = sql();
await db`
create table if not exists shared_app_state (
state_key text primary key,
payload jsonb not null,
updated_at timestamptz not null default now()
)
`;
sharedStateSchemaReady = true;
}
export async function readSharedAppState(stateKey) {
await ensureSharedStateSchema();
const db = sql();
const rows = await db`
select payload, updated_at
from shared_app_state
where state_key = ${stateKey}
limit 1
`;
return rows[0] || null;
}
export async function writeSharedAppState(stateKey, payload) {
await ensureSharedStateSchema();
const db = sql();
const rows = await db`
insert into shared_app_state (state_key, payload, updated_at)
values (${stateKey}, ${JSON.stringify(payload)}::jsonb, now())
on conflict (state_key) do update set
payload = excluded.payload,
updated_at = now()
returning updated_at
`;
return rows[0] || null;
}
export async function ensureSchema() { export async function ensureSchema() {
if (schemaReady) return; if (schemaReady) return;
const db = sql(); const db = sql();
+47 -9
View File
@@ -1,7 +1,9 @@
import { get, list, put } from "@vercel/blob"; import { get, list, put } from "@vercel/blob";
import { hasDatabase, readSharedAppState, writeSharedAppState } from "./_db.js";
const STATE_PATH = process.env.PMA_STATE_PATH || "shared/state.json"; const STATE_PATH = process.env.PMA_STATE_PATH || "shared/state.json";
const STATE_VERSION_PREFIX = process.env.PMA_STATE_VERSION_PREFIX || "shared/state-versions/"; const STATE_VERSION_PREFIX = process.env.PMA_STATE_VERSION_PREFIX || "shared/state-versions/";
const DATABASE_STATE_KEY = process.env.PMA_DATABASE_STATE_KEY || "polymarket-arena";
const AGENTS_KEY = "pma_agents_v2"; const AGENTS_KEY = "pma_agents_v2";
const SUG_KEY = "pma_suggestions_v5"; const SUG_KEY = "pma_suggestions_v5";
const PAPER_KEY = "pma_paper_accounts_v1"; const PAPER_KEY = "pma_paper_accounts_v1";
@@ -15,6 +17,17 @@ function withBlobAuth(options = {}) {
} }
async function readJsonBlob() { async function readJsonBlob() {
let databaseAvailable = false;
if (hasDatabase()) {
try {
const row = await readSharedAppState(DATABASE_STATE_KEY);
databaseAvailable = true;
if (row && row.payload) return row.payload;
} catch {
databaseAvailable = false;
}
}
let primaryError = null; let primaryError = null;
try { try {
const primary = await readBlobJson(STATE_PATH); const primary = await readBlobJson(STATE_PATH);
@@ -30,10 +43,40 @@ async function readJsonBlob() {
if (!primaryError) primaryError = err; if (!primaryError) primaryError = err;
} }
if (primaryError) throw primaryError; if (primaryError && !databaseAvailable) throw primaryError;
return null; return null;
} }
async function persistState(state) {
let databaseSaved = false;
let databaseError = null;
if (hasDatabase()) {
try {
await writeSharedAppState(DATABASE_STATE_KEY, state);
databaseSaved = true;
} catch (err) {
databaseError = err;
}
}
let blobSaved = false;
let blobError = null;
try {
await put(STATE_PATH, JSON.stringify(state), withBlobAuth({
access: "private",
allowOverwrite: true,
contentType: "application/json",
cacheControlMaxAge: 0,
}));
blobSaved = true;
} catch (err) {
blobError = err;
}
if (!databaseSaved && !blobSaved) throw databaseError || blobError || new Error("No state provider is available");
return { databaseSaved, blobSaved };
}
async function readBlobJson(pathname) { async function readBlobJson(pathname) {
const blob = await get(pathname, withBlobAuth({ const blob = await get(pathname, withBlobAuth({
access: "private", access: "private",
@@ -177,7 +220,7 @@ function conflictResponse(res, error, current) {
export default async function handler(req, res) { export default async function handler(req, res) {
res.setHeader("Cache-Control", "no-store"); res.setHeader("Cache-Control", "no-store");
try { try {
if (!process.env.BLOB_READ_WRITE_TOKEN && !process.env.VERCEL_OIDC_TOKEN) { if (!hasDatabase() && !process.env.BLOB_READ_WRITE_TOKEN && !process.env.VERCEL_OIDC_TOKEN) {
return res.status(503).json({ ok: false, error: "Cloud state is not configured" }); return res.status(503).json({ ok: false, error: "Cloud state is not configured" });
} }
@@ -223,13 +266,8 @@ export default async function handler(req, res) {
return conflictResponse(res, "Incoming state would replace newer active positions with stale cash-only data", current); return conflictResponse(res, "Incoming state would replace newer active positions with stale cash-only data", current);
} }
const state = { version: 1, updated_at: new Date().toISOString(), items: compactItems(incomingItems) }; const state = { version: 1, updated_at: new Date().toISOString(), items: compactItems(incomingItems) };
await put(STATE_PATH, JSON.stringify(state), withBlobAuth({ const saved = await persistState(state);
access: "private", if (saved.blobSaved && process.env.PMA_ENABLE_STATE_HISTORY === "true") {
allowOverwrite: true,
contentType: "application/json",
cacheControlMaxAge: 0,
}));
if (process.env.PMA_ENABLE_STATE_HISTORY === "true") {
const versionedPath = `${STATE_VERSION_PREFIX}${Date.now()}-${Math.random().toString(36).slice(2)}.json`; const versionedPath = `${STATE_VERSION_PREFIX}${Date.now()}-${Math.random().toString(36).slice(2)}.json`;
try { try {
await put(versionedPath, JSON.stringify(state), withBlobAuth({ await put(versionedPath, JSON.stringify(state), withBlobAuth({