Initial Commit
This commit is contained in:
@@ -0,0 +1,666 @@
|
||||
import {
|
||||
mutationGeneric,
|
||||
paginationOptsValidator,
|
||||
queryGeneric,
|
||||
} from "convex/server";
|
||||
import { v } from "convex/values";
|
||||
import { assetValidator } from "./schema";
|
||||
|
||||
const MAX_ASSET_CHUNK = 250;
|
||||
const SYNC_LEASE_MS = 5 * 60 * 1000;
|
||||
// Convex transactions must validate the manifest within bounded read limits.
|
||||
const MAX_CHUNKS = 4_096;
|
||||
const MAX_ERROR_LENGTH = 160;
|
||||
const OAUTH_MAX_LIFETIME_MS = 10 * 60 * 1000;
|
||||
const SNAPSHOT_MAX_LIFETIME_MS = 60 * 60 * 1000;
|
||||
|
||||
const bridgeArgs = { bridgeSecret: v.string() };
|
||||
const characterArgs = { userId: v.string(), characterId: v.number() };
|
||||
|
||||
const assetPageResultValidator = v.object({
|
||||
page: v.array(assetValidator),
|
||||
isDone: v.boolean(),
|
||||
continueCursor: v.string(),
|
||||
generation: v.union(v.string(), v.null()),
|
||||
});
|
||||
|
||||
function now(): number {
|
||||
return Date.now();
|
||||
}
|
||||
|
||||
/**
|
||||
* Bridge authentication is deliberately separate from Convex's deployment
|
||||
* admin key. The latter is only for operators and is never accepted here.
|
||||
*/
|
||||
function assertBridge(secret: string): void {
|
||||
const expected = process.env.CONVEX_BRIDGE_SECRET;
|
||||
if (!expected || !constantTimeEqual(secret, expected)) {
|
||||
throw new Error("Convex bridge authentication failed");
|
||||
}
|
||||
}
|
||||
|
||||
function constantTimeEqual(a: string, b: string): boolean {
|
||||
let difference = a.length ^ b.length;
|
||||
const length = Math.max(a.length, b.length);
|
||||
for (let index = 0; index < length; index += 1) {
|
||||
difference |= (a.charCodeAt(index) || 0) ^ (b.charCodeAt(index) || 0);
|
||||
}
|
||||
return difference === 0;
|
||||
}
|
||||
|
||||
/** Do not persist ESI/provider response text for a browser-facing status. */
|
||||
function safeSyncError(_error: string): string {
|
||||
return "Asset synchronization failed";
|
||||
}
|
||||
|
||||
function newIdentifier(prefix: string): string {
|
||||
return `${prefix}_${crypto.randomUUID()}`;
|
||||
}
|
||||
|
||||
async function findCharacter(ctx: any, userId: string, characterId: number): Promise<any | null> {
|
||||
return await ctx.db
|
||||
.query("characters")
|
||||
.withIndex("by_user_character", (q: any) =>
|
||||
q.eq("userId", userId).eq("characterId", characterId),
|
||||
)
|
||||
.first();
|
||||
}
|
||||
|
||||
async function findSessionByHash(ctx: any, sessionHash: string): Promise<any | null> {
|
||||
return await ctx.db
|
||||
.query("sessions")
|
||||
.withIndex("by_session_hash", (q: any) => q.eq("sessionHash", sessionHash))
|
||||
.first();
|
||||
}
|
||||
|
||||
function isUsableSession(session: any, timestamp = now()): boolean {
|
||||
return Boolean(
|
||||
session &&
|
||||
!session.revokedAt &&
|
||||
session.expiresAt > timestamp,
|
||||
);
|
||||
}
|
||||
|
||||
/** Server-only session lookup. The bridge secret is required even for reads. */
|
||||
export const bridgeSession = queryGeneric({
|
||||
args: { ...bridgeArgs, sessionHash: v.string() },
|
||||
handler: async (ctx, args) => {
|
||||
assertBridge(args.bridgeSecret);
|
||||
const session = await findSessionByHash(ctx, args.sessionHash);
|
||||
if (!isUsableSession(session)) return null;
|
||||
return { userId: session.userId, sessionId: session.sessionId };
|
||||
},
|
||||
});
|
||||
|
||||
/** Revocation is represented by a tombstone so JWT `sid` checks remain effective. */
|
||||
export const bridgeDeleteSession = mutationGeneric({
|
||||
args: { ...bridgeArgs, sessionHash: v.string() },
|
||||
handler: async (ctx, args) => {
|
||||
assertBridge(args.bridgeSecret);
|
||||
const session = await findSessionByHash(ctx, args.sessionHash);
|
||||
if (session && !session.revokedAt) {
|
||||
await ctx.db.patch(session._id, { revokedAt: now() });
|
||||
}
|
||||
return null;
|
||||
},
|
||||
});
|
||||
|
||||
export const bridgeCreateOAuth = mutationGeneric({
|
||||
args: {
|
||||
...bridgeArgs,
|
||||
stateHash: v.string(),
|
||||
bindingHash: v.string(),
|
||||
userId: v.optional(v.string()),
|
||||
encryptedVerifier: v.string(),
|
||||
expiresAt: v.number(),
|
||||
returnTo: v.string(),
|
||||
},
|
||||
handler: async (ctx, args) => {
|
||||
assertBridge(args.bridgeSecret);
|
||||
const timestamp = now();
|
||||
if (
|
||||
!Number.isFinite(args.expiresAt) ||
|
||||
args.expiresAt <= timestamp ||
|
||||
args.expiresAt > timestamp + OAUTH_MAX_LIFETIME_MS
|
||||
) {
|
||||
throw new Error("Invalid OAuth state expiry");
|
||||
}
|
||||
// Keep this bounded cleanup opportunistic; retention/GC is intentionally
|
||||
// outside this bridge transaction.
|
||||
const expired = await ctx.db
|
||||
.query("oauthStates")
|
||||
.withIndex("by_expires_at", (q: any) => q.lte("expiresAt", timestamp))
|
||||
.take(50);
|
||||
for (const state of expired) await ctx.db.delete(state._id);
|
||||
|
||||
const existing = await ctx.db
|
||||
.query("oauthStates")
|
||||
.withIndex("by_state_hash", (q: any) => q.eq("stateHash", args.stateHash))
|
||||
.first();
|
||||
if (existing) throw new Error("OAuth state already exists");
|
||||
await ctx.db.insert("oauthStates", {
|
||||
stateHash: args.stateHash,
|
||||
bindingHash: args.bindingHash,
|
||||
...(args.userId === undefined ? {} : { userId: args.userId }),
|
||||
encryptedVerifier: args.encryptedVerifier,
|
||||
expiresAt: args.expiresAt,
|
||||
returnTo: args.returnTo,
|
||||
});
|
||||
return null;
|
||||
},
|
||||
});
|
||||
|
||||
/** Consume and mark the state in one mutation; concurrent callbacks cannot reuse it. */
|
||||
export const bridgeConsumeOAuth = mutationGeneric({
|
||||
args: { ...bridgeArgs, stateHash: v.string(), bindingHash: v.string() },
|
||||
handler: async (ctx, args) => {
|
||||
assertBridge(args.bridgeSecret);
|
||||
const state = await ctx.db
|
||||
.query("oauthStates")
|
||||
.withIndex("by_state_binding", (q: any) =>
|
||||
q.eq("stateHash", args.stateHash).eq("bindingHash", args.bindingHash),
|
||||
)
|
||||
.first();
|
||||
if (!state || state.consumedAt !== undefined) return null;
|
||||
const consumedAt = now();
|
||||
await ctx.db.patch(state._id, { consumedAt });
|
||||
if (state.expiresAt <= consumedAt) return null;
|
||||
return {
|
||||
stateHash: state.stateHash,
|
||||
bindingHash: state.bindingHash,
|
||||
...(state.userId === undefined ? {} : { userId: state.userId }),
|
||||
encryptedVerifier: state.encryptedVerifier,
|
||||
expiresAt: state.expiresAt,
|
||||
returnTo: state.returnTo,
|
||||
};
|
||||
},
|
||||
});
|
||||
|
||||
const completeLoginArgs = {
|
||||
...bridgeArgs,
|
||||
userId: v.optional(v.string()),
|
||||
characterId: v.number(),
|
||||
name: v.string(),
|
||||
ownerHash: v.string(),
|
||||
scopes: v.array(v.string()),
|
||||
encryptedTokens: v.string(),
|
||||
tokenExpiresAt: v.number(),
|
||||
sessionHash: v.string(),
|
||||
sessionExpiresAt: v.number(),
|
||||
};
|
||||
|
||||
/**
|
||||
* Create/login/link a character and rotate its token record in one transaction.
|
||||
* An owner hash is continuity evidence for this character only; it is never
|
||||
* used to infer that two characters belong to the same EVE account.
|
||||
*/
|
||||
export const bridgeCompleteLogin = mutationGeneric({
|
||||
args: completeLoginArgs,
|
||||
handler: async (ctx, args) => {
|
||||
assertBridge(args.bridgeSecret);
|
||||
const byCharacterId = await ctx.db
|
||||
.query("characters")
|
||||
.withIndex("by_character_id", (q: any) => q.eq("characterId", args.characterId))
|
||||
.collect();
|
||||
if (byCharacterId.some((character: any) => character.ownerHash !== args.ownerHash)) {
|
||||
throw new Error("Character ownership conflict");
|
||||
}
|
||||
const existingCharacter = byCharacterId[0] ?? null;
|
||||
|
||||
let resolvedUserId: string;
|
||||
if (existingCharacter) {
|
||||
if (existingCharacter.ownerHash !== args.ownerHash) {
|
||||
throw new Error("Character ownership conflict");
|
||||
}
|
||||
if (args.userId !== undefined && args.userId !== existingCharacter.userId) {
|
||||
throw new Error("Character belongs to another user");
|
||||
}
|
||||
resolvedUserId = existingCharacter.userId;
|
||||
} else {
|
||||
// A signed-out first login creates one app user for this character. A
|
||||
// new character may only be attached when the current app user is
|
||||
// explicitly supplied by the authenticated linking flow.
|
||||
resolvedUserId = args.userId ?? newIdentifier("user");
|
||||
}
|
||||
|
||||
const sameSession = await findSessionByHash(ctx, args.sessionHash);
|
||||
if (sameSession) {
|
||||
throw new Error("Session hash collision");
|
||||
}
|
||||
|
||||
const user = await ctx.db
|
||||
.query("users")
|
||||
.withIndex("by_user_id", (q: any) => q.eq("userId", resolvedUserId))
|
||||
.first();
|
||||
if (!user) {
|
||||
if (args.userId !== undefined || existingCharacter) {
|
||||
throw new Error("User does not exist");
|
||||
}
|
||||
await ctx.db.insert("users", { userId: resolvedUserId, createdAt: now() });
|
||||
}
|
||||
|
||||
let character = existingCharacter;
|
||||
if (!character) {
|
||||
character = await findCharacter(ctx, resolvedUserId, args.characterId);
|
||||
}
|
||||
if (character) {
|
||||
if (character.ownerHash !== args.ownerHash || character.userId !== resolvedUserId) {
|
||||
throw new Error("Character ownership conflict");
|
||||
}
|
||||
await ctx.db.patch(character._id, {
|
||||
name: args.name,
|
||||
scopes: args.scopes,
|
||||
ownerHash: args.ownerHash,
|
||||
});
|
||||
} else {
|
||||
await ctx.db.insert("characters", {
|
||||
userId: resolvedUserId,
|
||||
characterId: args.characterId,
|
||||
name: args.name,
|
||||
ownerHash: args.ownerHash,
|
||||
scopes: args.scopes,
|
||||
state: "never",
|
||||
});
|
||||
}
|
||||
|
||||
const tokenRecord = await ctx.db
|
||||
.query("tokenRecords")
|
||||
.withIndex("by_user_character", (q: any) =>
|
||||
q.eq("userId", resolvedUserId).eq("characterId", args.characterId),
|
||||
)
|
||||
.first();
|
||||
if (tokenRecord) {
|
||||
await ctx.db.patch(tokenRecord._id, {
|
||||
encryptedTokens: args.encryptedTokens,
|
||||
tokenExpiresAt: args.tokenExpiresAt,
|
||||
scopes: args.scopes,
|
||||
version: tokenRecord.version + 1,
|
||||
});
|
||||
} else {
|
||||
await ctx.db.insert("tokenRecords", {
|
||||
userId: resolvedUserId,
|
||||
characterId: args.characterId,
|
||||
encryptedTokens: args.encryptedTokens,
|
||||
tokenExpiresAt: args.tokenExpiresAt,
|
||||
scopes: args.scopes,
|
||||
version: 1,
|
||||
});
|
||||
}
|
||||
|
||||
const sessionId = newIdentifier("session");
|
||||
await ctx.db.insert("sessions", {
|
||||
sessionHash: args.sessionHash,
|
||||
sessionId,
|
||||
userId: resolvedUserId,
|
||||
expiresAt: args.sessionExpiresAt,
|
||||
});
|
||||
return { userId: resolvedUserId, sessionId };
|
||||
},
|
||||
});
|
||||
|
||||
export const bridgeTokenRecord = queryGeneric({
|
||||
args: { ...bridgeArgs, ...characterArgs },
|
||||
handler: async (ctx, args) => {
|
||||
assertBridge(args.bridgeSecret);
|
||||
const character = await findCharacter(ctx, args.userId, args.characterId);
|
||||
if (!character) return null;
|
||||
return await ctx.db
|
||||
.query("tokenRecords")
|
||||
.withIndex("by_user_character", (q: any) =>
|
||||
q.eq("userId", args.userId).eq("characterId", args.characterId),
|
||||
)
|
||||
.first()
|
||||
.then((record: any) =>
|
||||
record
|
||||
? {
|
||||
encryptedTokens: record.encryptedTokens,
|
||||
tokenExpiresAt: record.tokenExpiresAt,
|
||||
version: record.version,
|
||||
scopes: record.scopes,
|
||||
}
|
||||
: null,
|
||||
);
|
||||
},
|
||||
});
|
||||
|
||||
export const bridgeSaveTokens = mutationGeneric({
|
||||
args: {
|
||||
...bridgeArgs,
|
||||
...characterArgs,
|
||||
encryptedTokens: v.string(),
|
||||
tokenExpiresAt: v.number(),
|
||||
scopes: v.array(v.string()),
|
||||
expectedVersion: v.number(),
|
||||
},
|
||||
handler: async (ctx, args) => {
|
||||
assertBridge(args.bridgeSecret);
|
||||
const character = await findCharacter(ctx, args.userId, args.characterId);
|
||||
if (!character) return false;
|
||||
const record = await ctx.db
|
||||
.query("tokenRecords")
|
||||
.withIndex("by_user_character", (q: any) =>
|
||||
q.eq("userId", args.userId).eq("characterId", args.characterId),
|
||||
)
|
||||
.first();
|
||||
if (!record || record.version !== args.expectedVersion) return false;
|
||||
await ctx.db.patch(record._id, {
|
||||
encryptedTokens: args.encryptedTokens,
|
||||
tokenExpiresAt: args.tokenExpiresAt,
|
||||
scopes: args.scopes,
|
||||
version: record.version + 1,
|
||||
});
|
||||
await ctx.db.patch(character._id, { scopes: args.scopes });
|
||||
return true;
|
||||
},
|
||||
});
|
||||
|
||||
export const bridgeCharacters = queryGeneric({
|
||||
args: { ...bridgeArgs, userId: v.string() },
|
||||
handler: async (ctx, args) => {
|
||||
assertBridge(args.bridgeSecret);
|
||||
return await ctx.db
|
||||
.query("characters")
|
||||
.withIndex("by_user_id", (q: any) => q.eq("userId", args.userId))
|
||||
.collect();
|
||||
},
|
||||
});
|
||||
|
||||
/**
|
||||
* Read an active snapshot one page of chunks at a time. The application adapter
|
||||
* walks this query's cursor, rather than issuing an unbounded Convex collect.
|
||||
*/
|
||||
export const bridgeAssetsPage = queryGeneric({
|
||||
args: {
|
||||
...bridgeArgs,
|
||||
...characterArgs,
|
||||
paginationOpts: paginationOptsValidator,
|
||||
expectedGeneration: v.optional(v.string()),
|
||||
},
|
||||
returns: assetPageResultValidator,
|
||||
handler: async (ctx, args) => {
|
||||
assertBridge(args.bridgeSecret);
|
||||
const character = await findCharacter(ctx, args.userId, args.characterId);
|
||||
const generation = character?.activeGeneration ?? null;
|
||||
if (!character || !generation) {
|
||||
return { page: [], isDone: true, continueCursor: "", generation: null };
|
||||
}
|
||||
// The caller pins the first page's generation. Returning the new generation
|
||||
// without its data prevents a read from silently mixing two snapshots.
|
||||
if (args.expectedGeneration !== undefined && args.expectedGeneration !== generation) {
|
||||
return { page: [], isDone: true, continueCursor: "", generation };
|
||||
}
|
||||
const page = await ctx.db
|
||||
.query("assetChunks")
|
||||
.withIndex("by_snapshot_index", (q: any) =>
|
||||
q
|
||||
.eq("userId", args.userId)
|
||||
.eq("characterId", args.characterId)
|
||||
.eq("generation", generation),
|
||||
)
|
||||
.order("asc")
|
||||
.paginate(args.paginationOpts);
|
||||
return {
|
||||
page: page.page.flatMap((chunk: any) => chunk.assets),
|
||||
isDone: page.isDone,
|
||||
continueCursor: page.continueCursor,
|
||||
generation,
|
||||
};
|
||||
},
|
||||
});
|
||||
|
||||
export const bridgeBeginSync = mutationGeneric({
|
||||
args: { ...bridgeArgs, ...characterArgs },
|
||||
handler: async (ctx, args) => {
|
||||
assertBridge(args.bridgeSecret);
|
||||
const character = await findCharacter(ctx, args.userId, args.characterId);
|
||||
if (!character) throw new Error("Character not found");
|
||||
const timestamp = now();
|
||||
if (character.syncGeneration && (character.syncLeaseExpiresAt ?? 0) > timestamp) {
|
||||
throw new Error("Character synchronization is already leased");
|
||||
}
|
||||
const generation = newIdentifier("generation");
|
||||
await ctx.db.insert("assetGenerations", {
|
||||
userId: args.userId,
|
||||
characterId: args.characterId,
|
||||
generation,
|
||||
status: "staging",
|
||||
stagedChunkCount: 0,
|
||||
});
|
||||
await ctx.db.patch(character._id, {
|
||||
state: "syncing",
|
||||
syncGeneration: generation,
|
||||
syncLeaseExpiresAt: timestamp + SYNC_LEASE_MS,
|
||||
error: undefined,
|
||||
});
|
||||
return generation;
|
||||
},
|
||||
});
|
||||
|
||||
export const bridgeStageAssets = mutationGeneric({
|
||||
args: {
|
||||
...bridgeArgs,
|
||||
...characterArgs,
|
||||
generation: v.string(),
|
||||
index: v.number(),
|
||||
assets: v.array(assetValidator),
|
||||
},
|
||||
handler: async (ctx, args) => {
|
||||
assertBridge(args.bridgeSecret);
|
||||
if (!Number.isInteger(args.index) || args.index < 0 || args.assets.length > MAX_ASSET_CHUNK) {
|
||||
throw new Error("Invalid asset chunk");
|
||||
}
|
||||
if (args.assets.some((asset: any) => asset.characterId !== args.characterId)) {
|
||||
throw new Error("Asset character mismatch");
|
||||
}
|
||||
const character = await findCharacter(ctx, args.userId, args.characterId);
|
||||
if (
|
||||
!character ||
|
||||
character.syncGeneration !== args.generation ||
|
||||
(character.syncLeaseExpiresAt ?? 0) <= now()
|
||||
) {
|
||||
throw new Error("Synchronization lease is not valid");
|
||||
}
|
||||
const generation = await ctx.db
|
||||
.query("assetGenerations")
|
||||
.withIndex("by_user_character_generation", (q: any) =>
|
||||
q
|
||||
.eq("userId", args.userId)
|
||||
.eq("characterId", args.characterId)
|
||||
.eq("generation", args.generation),
|
||||
)
|
||||
.first();
|
||||
if (!generation || generation.status !== "staging") throw new Error("Snapshot is not staging");
|
||||
|
||||
const stagedChunkCount = generation.stagedChunkCount ?? 0;
|
||||
if (stagedChunkCount >= MAX_CHUNKS || args.index !== stagedChunkCount) {
|
||||
throw new Error("Asset chunks must be staged exactly once and in order");
|
||||
}
|
||||
const existing = await ctx.db
|
||||
.query("assetChunks")
|
||||
.withIndex("by_snapshot_index", (q: any) =>
|
||||
q
|
||||
.eq("userId", args.userId)
|
||||
.eq("characterId", args.characterId)
|
||||
.eq("generation", args.generation)
|
||||
.eq("index", args.index),
|
||||
)
|
||||
.first();
|
||||
if (existing) throw new Error("Asset chunk already exists");
|
||||
// Insert and counter advancement are one mutation. A failed insert rolls
|
||||
// back the counter, while a duplicate/out-of-order index is rejected.
|
||||
await ctx.db.insert("assetChunks", {
|
||||
userId: args.userId,
|
||||
characterId: args.characterId,
|
||||
generation: args.generation,
|
||||
index: args.index,
|
||||
assets: args.assets,
|
||||
});
|
||||
await ctx.db.patch(generation._id, { stagedChunkCount: stagedChunkCount + 1 });
|
||||
return null;
|
||||
},
|
||||
});
|
||||
|
||||
export const bridgeFinishSync = mutationGeneric({
|
||||
args: {
|
||||
...bridgeArgs,
|
||||
...characterArgs,
|
||||
generation: v.string(),
|
||||
chunkCount: v.number(),
|
||||
syncedAt: v.number(),
|
||||
expiresAt: v.number(),
|
||||
},
|
||||
handler: async (ctx, args) => {
|
||||
assertBridge(args.bridgeSecret);
|
||||
const timestamp = now();
|
||||
if (
|
||||
!Number.isFinite(args.syncedAt) ||
|
||||
args.syncedAt < 0 ||
|
||||
args.syncedAt > timestamp ||
|
||||
!Number.isFinite(args.expiresAt) ||
|
||||
args.expiresAt <= timestamp ||
|
||||
args.expiresAt <= args.syncedAt ||
|
||||
args.expiresAt - args.syncedAt > SNAPSHOT_MAX_LIFETIME_MS
|
||||
) {
|
||||
throw new Error("Invalid snapshot timestamps");
|
||||
}
|
||||
if (!Number.isInteger(args.chunkCount) || args.chunkCount < 0 || args.chunkCount > MAX_CHUNKS) {
|
||||
throw new Error("Invalid snapshot chunk count");
|
||||
}
|
||||
const character = await findCharacter(ctx, args.userId, args.characterId);
|
||||
if (
|
||||
!character ||
|
||||
character.syncGeneration !== args.generation ||
|
||||
(character.syncLeaseExpiresAt ?? 0) <= now()
|
||||
) {
|
||||
throw new Error("Synchronization lease is not valid");
|
||||
}
|
||||
const generation = await ctx.db
|
||||
.query("assetGenerations")
|
||||
.withIndex("by_user_character_generation", (q: any) =>
|
||||
q
|
||||
.eq("userId", args.userId)
|
||||
.eq("characterId", args.characterId)
|
||||
.eq("generation", args.generation),
|
||||
)
|
||||
.first();
|
||||
if (!generation || generation.status !== "staging") throw new Error("Snapshot is not staging");
|
||||
|
||||
// Staging maintains a contiguous durable counter, so publication can
|
||||
// validate completeness without reading every asset payload.
|
||||
if ((generation.stagedChunkCount ?? 0) !== args.chunkCount) {
|
||||
throw new Error("Snapshot is missing chunks");
|
||||
}
|
||||
|
||||
await ctx.db.patch(generation._id, {
|
||||
status: "complete",
|
||||
chunkCount: args.chunkCount,
|
||||
syncedAt: args.syncedAt,
|
||||
expiresAt: args.expiresAt,
|
||||
});
|
||||
await ctx.db.patch(character._id, {
|
||||
state: "ready",
|
||||
syncedAt: args.syncedAt,
|
||||
expiresAt: args.expiresAt,
|
||||
activeGeneration: args.generation,
|
||||
syncGeneration: undefined,
|
||||
syncLeaseExpiresAt: undefined,
|
||||
error: undefined,
|
||||
});
|
||||
return null;
|
||||
},
|
||||
});
|
||||
|
||||
export const bridgeFailSync = mutationGeneric({
|
||||
args: { ...bridgeArgs, ...characterArgs, generation: v.string(), error: v.string() },
|
||||
handler: async (ctx, args) => {
|
||||
assertBridge(args.bridgeSecret);
|
||||
const character = await findCharacter(ctx, args.userId, args.characterId);
|
||||
if (!character || character.syncGeneration !== args.generation) {
|
||||
throw new Error("Synchronization generation is not current");
|
||||
}
|
||||
const generation = await ctx.db
|
||||
.query("assetGenerations")
|
||||
.withIndex("by_user_character_generation", (q: any) =>
|
||||
q
|
||||
.eq("userId", args.userId)
|
||||
.eq("characterId", args.characterId)
|
||||
.eq("generation", args.generation),
|
||||
)
|
||||
.first();
|
||||
if (generation) {
|
||||
await ctx.db.patch(generation._id, {
|
||||
status: "failed",
|
||||
error: safeSyncError(args.error).slice(0, MAX_ERROR_LENGTH),
|
||||
});
|
||||
}
|
||||
// Deliberately do not touch activeGeneration: the previous complete
|
||||
// snapshot remains available after a failed synchronization.
|
||||
await ctx.db.patch(character._id, {
|
||||
state: "error",
|
||||
error: safeSyncError(args.error).slice(0, MAX_ERROR_LENGTH),
|
||||
syncGeneration: undefined,
|
||||
syncLeaseExpiresAt: undefined,
|
||||
});
|
||||
return null;
|
||||
},
|
||||
});
|
||||
|
||||
export const bridgeSaveInteraction = mutationGeneric({
|
||||
args: {
|
||||
...bridgeArgs,
|
||||
userId: v.string(),
|
||||
requestId: v.string(),
|
||||
message: v.string(),
|
||||
result: v.any(),
|
||||
status: v.string(),
|
||||
},
|
||||
handler: async (ctx, args) => {
|
||||
assertBridge(args.bridgeSecret);
|
||||
const existing = await ctx.db
|
||||
.query("interactions")
|
||||
.withIndex("by_user_request", (q: any) =>
|
||||
q.eq("userId", args.userId).eq("requestId", args.requestId),
|
||||
)
|
||||
.first();
|
||||
if (existing) {
|
||||
if (existing.message !== args.message) {
|
||||
throw new Error("Interaction request ID was reused with a different message");
|
||||
}
|
||||
// A request can first be durably recorded as blocked by auth or missing
|
||||
// sync requirements, then resumed after the user completes that action.
|
||||
// Keep the idempotency key and original message, but advance its result.
|
||||
if (existing.status !== args.status || existing.result !== args.result) {
|
||||
await ctx.db.patch(existing._id, {
|
||||
result: args.result,
|
||||
status: args.status,
|
||||
});
|
||||
}
|
||||
return {
|
||||
requestId: existing.requestId,
|
||||
message: existing.message,
|
||||
result: args.result,
|
||||
status: args.status,
|
||||
createdAt: existing.createdAt,
|
||||
created: false,
|
||||
};
|
||||
}
|
||||
const createdAt = now();
|
||||
await ctx.db.insert("interactions", {
|
||||
userId: args.userId,
|
||||
requestId: args.requestId,
|
||||
message: args.message,
|
||||
result: args.result,
|
||||
status: args.status,
|
||||
createdAt,
|
||||
});
|
||||
return {
|
||||
requestId: args.requestId,
|
||||
message: args.message,
|
||||
result: args.result,
|
||||
status: args.status,
|
||||
createdAt,
|
||||
created: true,
|
||||
};
|
||||
},
|
||||
});
|
||||
|
||||
Reference in New Issue
Block a user