405 lines
17 KiB
TypeScript
405 lines
17 KiB
TypeScript
import type { Db } from "@sttracker/db/db";
|
|
import { newId, nowIso } from "@sttracker/db/ids";
|
|
import { fromJson, toJson } from "@sttracker/db/json";
|
|
import type { ConnectionStatus, RunMode, RunStatus } from "@sttracker/db/schema";
|
|
import { createHttpClient, type HttpClientOptions } from "@sttracker/plugin-sdk/http";
|
|
import type { HttpClient, PluginContext, PluginManifest } from "@sttracker/plugin-sdk/types";
|
|
import { decryptJson, encryptJson } from "./crypto.ts";
|
|
import { CryptoError, NotFoundError, ValidationError } from "./errors.ts";
|
|
import { DropJobError, NonRetryableError, type JobContext } from "./jobs.ts";
|
|
import type { PluginRegistry } from "./registry.ts";
|
|
import { runSync, type Connection, type RunSummary, type SyncMode } from "./sync/run.ts";
|
|
|
|
const WEEK_MS = 7 * 24 * 60 * 60 * 1000;
|
|
|
|
export interface ConnectionView {
|
|
id: string;
|
|
pluginId: string;
|
|
name: string;
|
|
config: Record<string, unknown>;
|
|
scheduleMinutes: number | null;
|
|
status: ConnectionStatus;
|
|
lastRunAt: string | null;
|
|
createdAt: string;
|
|
hasCredentials: boolean;
|
|
}
|
|
|
|
export interface ConnectionInput {
|
|
pluginId: string;
|
|
name: string;
|
|
config: Record<string, unknown>;
|
|
credentials?: Record<string, string>;
|
|
scheduleMinutes?: number | null;
|
|
}
|
|
|
|
type Creds = Record<string, string>;
|
|
|
|
/** Split incoming config: fields typed `"secret"` in the manifest go to the credentials blob, the rest stay config. */
|
|
function splitConfig(manifest: PluginManifest, config: Record<string, unknown>): { plain: Record<string, unknown>; secrets: Creds } {
|
|
const secretKeys = new Set(manifest.config.filter((f) => f.type === "secret").map((f) => f.key));
|
|
const plain: Record<string, unknown> = {};
|
|
const secrets: Creds = {};
|
|
for (const [k, v] of Object.entries(config)) {
|
|
if (!secretKeys.has(k)) plain[k] = v;
|
|
else if (typeof v === "string" && v !== "") secrets[k] = v;
|
|
else if (v !== undefined && v !== null && v !== "") throw new ValidationError(k, "secret_must_be_string");
|
|
}
|
|
return { plain, secrets };
|
|
}
|
|
|
|
async function readCreds(secret: string, enc: string | null): Promise<Creds | null> {
|
|
return enc === null ? null : decryptJson<Creds>(secret, enc);
|
|
}
|
|
|
|
const encOrNull = async (secret: string, c: Creds): Promise<string | null> => (Object.keys(c).length === 0 ? null : encryptJson(secret, c));
|
|
|
|
function checkSchedule(m: number | null | undefined): void {
|
|
if (m !== undefined && m !== null && (!Number.isInteger(m) || m < 1)) throw new ValidationError("scheduleMinutes", "must_be_positive_integer");
|
|
}
|
|
|
|
export async function createConnection(db: Db, secret: string, userId: string, input: ConnectionInput, manifest: PluginManifest): Promise<string> {
|
|
if (manifest.id !== input.pluginId) throw new ValidationError("pluginId", "manifest_mismatch");
|
|
if (input.name.trim() === "") throw new ValidationError("name", "required");
|
|
checkSchedule(input.scheduleMinutes);
|
|
const { plain, secrets } = splitConfig(manifest, input.config);
|
|
const id = newId();
|
|
await db.k
|
|
.insertInto("source_connections")
|
|
.values({
|
|
id,
|
|
user_id: userId,
|
|
plugin_id: input.pluginId,
|
|
name: input.name.trim(),
|
|
config: toJson(plain),
|
|
credentials_enc: await encOrNull(secret, { ...secrets, ...input.credentials }),
|
|
schedule_minutes: input.scheduleMinutes ?? null,
|
|
cursor: null,
|
|
status: "active",
|
|
last_run_at: null,
|
|
created_at: nowIso(),
|
|
})
|
|
.execute();
|
|
return id;
|
|
}
|
|
|
|
export interface ConnectionPatch {
|
|
name?: string;
|
|
config?: Record<string, unknown>;
|
|
credentials?: Record<string, string>;
|
|
scheduleMinutes?: number | null;
|
|
status?: ConnectionStatus;
|
|
}
|
|
|
|
/** Secrets omitted from `config` are kept; plain config is replaced wholesale. Read-modify-write runs in one transaction. */
|
|
export async function updateConnection(db: Db, secret: string, userId: string, id: string, patch: ConnectionPatch, manifest: PluginManifest): Promise<void> {
|
|
checkSchedule(patch.scheduleMinutes);
|
|
await db.k.transaction().execute(async (tx) => {
|
|
let q = tx.selectFrom("source_connections").selectAll().where("id", "=", id).where("user_id", "=", userId);
|
|
if (db.dialect === "postgres") q = q.forUpdate();
|
|
const row = await q.executeTakeFirst();
|
|
if (!row) throw new NotFoundError("connection");
|
|
if (row.plugin_id !== manifest.id) throw new ValidationError("pluginId", "manifest_mismatch");
|
|
const set: Record<string, unknown> = {};
|
|
if (patch.name !== undefined) {
|
|
if (patch.name.trim() === "") throw new ValidationError("name", "required");
|
|
set.name = patch.name.trim();
|
|
}
|
|
if (patch.scheduleMinutes !== undefined) set.schedule_minutes = patch.scheduleMinutes;
|
|
if (patch.status !== undefined) set.status = patch.status;
|
|
if (patch.config !== undefined || patch.credentials !== undefined) {
|
|
const split = patch.config ? splitConfig(manifest, patch.config) : { plain: undefined, secrets: {} };
|
|
if (split.plain) set.config = toJson(split.plain);
|
|
const merged = { ...((await readCreds(secret, row.credentials_enc)) ?? {}), ...split.secrets, ...patch.credentials };
|
|
set.credentials_enc = await encOrNull(secret, merged);
|
|
}
|
|
if (Object.keys(set).length > 0) await tx.updateTable("source_connections").set(set).where("id", "=", id).execute();
|
|
});
|
|
}
|
|
|
|
/** Soft delete: the connection is paused and its credentials are wiped. Records and runs stay. */
|
|
export async function deleteConnection(db: Db, userId: string, id: string): Promise<void> {
|
|
const r = await db.k
|
|
.updateTable("source_connections")
|
|
.set({ status: "paused", credentials_enc: null })
|
|
.where("id", "=", id)
|
|
.where("user_id", "=", userId)
|
|
.executeTakeFirst();
|
|
if (r.numUpdatedRows !== 1n) throw new NotFoundError("connection");
|
|
}
|
|
|
|
/** Never returns secrets: plain config plus `hasCredentials`. */
|
|
export async function listConnections(db: Db, userId: string): Promise<ConnectionView[]> {
|
|
const rows = await db.k.selectFrom("source_connections").selectAll().where("user_id", "=", userId).orderBy("created_at").execute();
|
|
return rows.map((r) => ({
|
|
id: r.id,
|
|
pluginId: r.plugin_id,
|
|
name: r.name,
|
|
config: fromJson<Record<string, unknown>>(r.config, {}),
|
|
scheduleMinutes: r.schedule_minutes,
|
|
status: r.status,
|
|
lastRunAt: r.last_run_at,
|
|
createdAt: r.created_at,
|
|
hasCredentials: r.credentials_enc !== null,
|
|
}));
|
|
}
|
|
|
|
/** A deleted connection is paused with its credentials wiped (`deleteConnection`). */
|
|
export function isDeletedConnection(c: Pick<ConnectionView, "status" | "hasCredentials">): boolean {
|
|
return c.status === "paused" && !c.hasCredentials;
|
|
}
|
|
|
|
async function ownedRow(db: Db, userId: string, id: string) {
|
|
const r = await db.k.selectFrom("source_connections").selectAll().where("id", "=", id).where("user_id", "=", userId).executeTakeFirst();
|
|
if (!r || (r.status === "paused" && r.credentials_enc === null)) throw new NotFoundError("connection");
|
|
return r;
|
|
}
|
|
|
|
/** One of the user's live connections, without secrets. Deleted or foreign connections throw `NotFoundError`. */
|
|
export async function getConnection(db: Db, userId: string, id: string): Promise<ConnectionView> {
|
|
const r = await ownedRow(db, userId, id);
|
|
return {
|
|
id: r.id,
|
|
pluginId: r.plugin_id,
|
|
name: r.name,
|
|
config: fromJson<Record<string, unknown>>(r.config, {}),
|
|
scheduleMinutes: r.schedule_minutes,
|
|
status: r.status,
|
|
lastRunAt: r.last_run_at,
|
|
createdAt: r.created_at,
|
|
hasCredentials: r.credentials_enc !== null,
|
|
};
|
|
}
|
|
|
|
export interface RunView {
|
|
id: string;
|
|
mode: RunMode;
|
|
status: RunStatus;
|
|
startedAt: string;
|
|
finishedAt: string | null;
|
|
counts: Record<string, number>;
|
|
log: string;
|
|
error: string | null;
|
|
}
|
|
|
|
/** Runs of one of the user's connections, newest first (at most `limit`, default 50). */
|
|
export async function listRuns(db: Db, userId: string, connectionId: string, limit = 50): Promise<RunView[]> {
|
|
await ownedRow(db, userId, connectionId);
|
|
const rows = await db.k
|
|
.selectFrom("source_runs")
|
|
.selectAll()
|
|
.where("connection_id", "=", connectionId)
|
|
.orderBy("started_at", "desc")
|
|
.orderBy("id", "desc")
|
|
.limit(Math.min(Math.max(limit, 1), 200))
|
|
.execute();
|
|
return rows.map((r) => ({
|
|
id: r.id,
|
|
mode: r.mode,
|
|
status: r.status,
|
|
startedAt: r.started_at,
|
|
finishedAt: r.finished_at,
|
|
counts: fromJson<Record<string, number>>(r.counts, {}),
|
|
log: r.log,
|
|
error: r.error,
|
|
}));
|
|
}
|
|
|
|
async function deviceContext(db: Db, secret: string, registry: PluginRegistry, userId: string, id: string) {
|
|
const r = await ownedRow(db, userId, id);
|
|
const plugin = registry.get(r.plugin_id);
|
|
if (!plugin) throw new ValidationError("pluginId", "not_installed");
|
|
if (!plugin.deviceAuth) throw new ValidationError("pluginId", "no_device_auth");
|
|
const conn: Connection = {
|
|
id: r.id,
|
|
userId: r.user_id,
|
|
pluginId: r.plugin_id,
|
|
name: r.name,
|
|
config: fromJson<Record<string, unknown>>(r.config, {}),
|
|
credentialsEnc: r.credentials_enc,
|
|
scheduleMinutes: r.schedule_minutes,
|
|
cursor: r.cursor,
|
|
status: r.status,
|
|
lastRunAt: r.last_run_at,
|
|
createdAt: r.created_at,
|
|
};
|
|
const ctx = await makePluginContext(db, secret, conn, "device", plugin.manifest);
|
|
return { deviceAuth: plugin.deviceAuth, ctx };
|
|
}
|
|
|
|
/** Start a device-code sign-in for one of the user's connections. */
|
|
export async function deviceStart(db: Db, secret: string, registry: PluginRegistry, userId: string, id: string) {
|
|
const { deviceAuth, ctx } = await deviceContext(db, secret, registry, userId, id);
|
|
return deviceAuth.start(ctx);
|
|
}
|
|
|
|
/** Poll a device-code sign-in. On success the returned credentials are merged into the encrypted blob. */
|
|
export async function devicePoll(db: Db, secret: string, registry: PluginRegistry, userId: string, id: string, deviceCode: string): Promise<"pending" | "connected"> {
|
|
const { deviceAuth, ctx } = await deviceContext(db, secret, registry, userId, id);
|
|
const res = await deviceAuth.poll(ctx, deviceCode);
|
|
if (res === "pending") return "pending";
|
|
await ctx.credentials.save(res);
|
|
return "connected";
|
|
}
|
|
|
|
export const USER_AGENT = "STTracker/0.1 (+self-hosted)";
|
|
|
|
export interface ContextOpts {
|
|
/** Replaces the rate-limited client entirely. */
|
|
http?: HttpClient;
|
|
/** Overrides for the built client (fetch, sleep, clock); used by tests. */
|
|
httpOptions?: Pick<HttpClientOptions, "fetchImpl" | "sleep" | "now">;
|
|
signal?: AbortSignal;
|
|
/** Receives `ctx.log` messages. */
|
|
onLog?: (msg: string) => void;
|
|
}
|
|
|
|
/** Synchronous context builder over already-decrypted credentials. */
|
|
function buildContext(db: Db, secret: string, conn: Connection, manifest: PluginManifest, initial: Creds | null, opts: ContextOpts): PluginContext {
|
|
let creds = initial;
|
|
let cursor = conn.cursor;
|
|
return {
|
|
http:
|
|
opts.http ??
|
|
createHttpClient({
|
|
ratePerMinute: manifest.rateLimit?.perMinute ?? 60,
|
|
burst: manifest.rateLimit?.burst ?? 5,
|
|
userAgent: USER_AGENT,
|
|
...opts.httpOptions,
|
|
}),
|
|
log: (msg) => opts.onLog?.(msg),
|
|
config: { ...conn.config, ...(initial ?? {}) },
|
|
credentials: {
|
|
get: async () => creds,
|
|
async save(c) {
|
|
// Read the current blob, merge, and write in one transaction so concurrent saves cannot lose keys.
|
|
creds = await db.k.transaction().execute(async (tx) => {
|
|
let q = tx.selectFrom("source_connections").select("credentials_enc").where("id", "=", conn.id);
|
|
if (db.dialect === "postgres") q = q.forUpdate();
|
|
const row = await q.executeTakeFirstOrThrow();
|
|
const merged = { ...((await readCreds(secret, row.credentials_enc)) ?? {}), ...c };
|
|
await tx.updateTable("source_connections").set({ credentials_enc: await encryptJson(secret, merged) }).where("id", "=", conn.id).execute();
|
|
return merged;
|
|
});
|
|
},
|
|
},
|
|
cursor: {
|
|
get: async () => cursor,
|
|
async set(c) {
|
|
await db.k.updateTable("source_connections").set({ cursor: c }).where("id", "=", conn.id).execute();
|
|
cursor = c;
|
|
},
|
|
},
|
|
now: () => new Date(),
|
|
signal: opts.signal ?? new AbortController().signal,
|
|
};
|
|
}
|
|
|
|
/**
|
|
* Build the plugin context for one run. `ctx.http` is rate limited from `manifest.rateLimit` (default 60/min, burst 5)
|
|
* unless `opts.http` is given. `credentials.save` merges into the current blob and persists it encrypted immediately;
|
|
* `cursor.set` persists immediately.
|
|
*/
|
|
export async function makePluginContext(db: Db, secret: string, conn: Connection, _runId: string, manifest: PluginManifest, opts: ContextOpts = {}): Promise<PluginContext> {
|
|
return buildContext(db, secret, conn, manifest, await readCreds(secret, conn.credentialsEnc), opts);
|
|
}
|
|
|
|
export interface SyncHandlerDeps {
|
|
db: Db;
|
|
secret: string;
|
|
registry: PluginRegistry;
|
|
http?: HttpClient;
|
|
}
|
|
|
|
const MSG_CRYPTO = "Cannot decrypt this source's credentials. Was STTRACKER_SECRET changed? Reconnect the source to fix it.";
|
|
|
|
/** Record a run that failed before `runSync` could start, flag the connection, and stop the job without retry. */
|
|
async function failBeforeRun(db: Db, connectionId: string, mode: RunMode, message: string): Promise<never> {
|
|
const now = nowIso();
|
|
await db.k
|
|
.insertInto("source_runs")
|
|
.values({ id: newId(), connection_id: connectionId, mode, started_at: now, finished_at: now, status: "failed", counts: "{}", log: message, error: message })
|
|
.execute();
|
|
await db.k.updateTable("source_connections").set({ status: "error", last_run_at: now }).where("id", "=", connectionId).execute();
|
|
throw new NonRetryableError(message);
|
|
}
|
|
|
|
/**
|
|
* The `"sync"` job handler. Payload `{ connectionId, mode? }`. Runs `full` when asked, when the plugin cannot
|
|
* `pullChanges`, or when the last successful full run is over a week old; otherwise `changes`.
|
|
* A missing or paused connection drops the job. A missing plugin or undecryptable credentials record a failed
|
|
* run, mark the connection `error` and fail the job without retry. A failed run is rethrown so the worker retries.
|
|
* `ctx.log` output is appended to the run's `source_runs.log`.
|
|
*/
|
|
export interface SyncPayload {
|
|
connectionId: string;
|
|
mode?: "full" | "changes";
|
|
/** File mode: run the plugin's `importFile` over these bytes instead of pulling. */
|
|
file?: { name: string; base64: string };
|
|
}
|
|
|
|
export function syncHandler(deps: SyncHandlerDeps): (payload: SyncPayload, jc?: JobContext) => Promise<RunSummary | void> {
|
|
return async (payload, jc) => {
|
|
const { db } = deps;
|
|
const full = await db.k.selectFrom("source_connections").selectAll().where("id", "=", payload.connectionId).executeTakeFirst();
|
|
if (!full) throw new DropJobError("connection not found");
|
|
if (full.status === "paused") throw new DropJobError("connection is paused");
|
|
const plugin = deps.registry.get(full.plugin_id);
|
|
if (!plugin) return failBeforeRun(db, full.id, payload.file ? "file" : (payload.mode ?? "full"), `Plugin '${full.plugin_id}' is not installed or failed to load.`);
|
|
let mode: SyncMode = "full";
|
|
if (payload.file) {
|
|
mode = { file: { name: payload.file.name, bytes: new Uint8Array(Buffer.from(payload.file.base64, "base64")) } };
|
|
} else if (plugin.pullChanges && payload.mode !== "full") {
|
|
const cutoff = new Date(Date.now() - WEEK_MS).toISOString();
|
|
const recentFull = await db.k
|
|
.selectFrom("source_runs")
|
|
.select("id")
|
|
.where("connection_id", "=", full.id)
|
|
.where("mode", "=", "full")
|
|
.where("status", "=", "ok")
|
|
.where("started_at", ">=", cutoff)
|
|
.executeTakeFirst();
|
|
if (recentFull) mode = "changes";
|
|
}
|
|
const conn: Connection = {
|
|
id: full.id,
|
|
userId: full.user_id,
|
|
pluginId: full.plugin_id,
|
|
name: full.name,
|
|
config: fromJson<Record<string, unknown>>(full.config, {}),
|
|
credentialsEnc: full.credentials_enc,
|
|
scheduleMinutes: full.schedule_minutes,
|
|
cursor: full.cursor,
|
|
status: full.status,
|
|
lastRunAt: full.last_run_at,
|
|
createdAt: full.created_at,
|
|
};
|
|
let creds: Creds | null;
|
|
try {
|
|
creds = await readCreds(deps.secret, conn.credentialsEnc);
|
|
} catch (e) {
|
|
if (e instanceof CryptoError) return failBeforeRun(db, full.id, typeof mode === "string" ? mode : "file", MSG_CRYPTO);
|
|
throw e;
|
|
}
|
|
const logs: string[] = [];
|
|
let runId: string | null = null;
|
|
const makeContext = (c: Connection, id: string) => {
|
|
runId = id;
|
|
return buildContext(db, deps.secret, c, plugin.manifest, creds, { http: deps.http, signal: jc?.signal, onLog: (m) => void logs.push(m) });
|
|
};
|
|
let summary: RunSummary;
|
|
try {
|
|
summary = await runSync({ db, makeContext }, plugin, full.id, mode);
|
|
} finally {
|
|
if (runId !== null && logs.length > 0) {
|
|
const cur = await db.k.selectFrom("source_runs").select("log").where("id", "=", runId).executeTakeFirst();
|
|
if (cur) {
|
|
const log = cur.log === "" ? logs.join("\n") : `${logs.join("\n")}\n${cur.log}`;
|
|
await db.k.updateTable("source_runs").set({ log }).where("id", "=", runId).execute();
|
|
}
|
|
}
|
|
}
|
|
if (summary.status === "failed") throw new Error(summary.error ?? "sync failed");
|
|
return summary;
|
|
};
|
|
}
|