rework: change backend and backend extensions to use simpler code structure and patterns (#2815)

* fix: dynamodb health checks and client recreation (#2789) * wip: no nanoServices groundwork * feat: data clients in new shape * wip: auth and perms in new system * more wip * middlewaters mainly done * wip: fsv2 in new layout * old fs v2 migration * driver system * driver and old fs fixes * ai drivers wip * stream support * metering in ai chat driver * wip: new auth * rate limit and auth routes * captcha and anti csrf * fix: types * auth store * app logic * wip most other dricvers * fs * mostly kill all legacy stuff * fs finish * fix: redis usage * ai controller * driver cleanup * socket io in v2 * broadcast and crudq stuff * subdomains * notifcations and shares * fix bad syntaxes * auth wip Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com> * extensions * extension setup * more routes * sql migrations and default services * home router * tier 7 * everything else * everything else * remaining missing bits * server health * logs * cleanup * deps * cleanup 2 * more cleanup 2 * boot * fix launch * config fix * move file * fix: tsconfig things * fix: extension loading * launching * fix: drivers * fix: others * fix: icons * fix: file uploads * fs fixes * fix: fs api * fix: dev-center * config * add back telemetry * lint stuff * husky hooks * fix: fs oss * fix: config migration * config migration * migrate scripts + replicate * runner * fix: merge defafult config * fix: default region * fix: api domain * fix paths in readfile * fix fs entry default s3 * NS: Remove Referral && Entri Service * dep cleanups * fix: static assets * fix: kv and perms * fix: driver registrations * fix: home mapping * fix: rao * adding back 500 alarm * fix: build paths * fix: fs and kv shapes * fix: kv shape * more kv coercing and ai chat matching format as prior * fix: private app gates * private app caches * fix: whole bunch of legacy shape issues * update template jsonc * fix caching partial oidc and fs signed paths * more oidc fixes * fix: wip * fix: private apps * admin route fixes * fix: last few things hopefully * claude uploads * fix security for app only routes * fix kv system namespace * stuff * fix: app and kv and suggested apps * fix:open item * fix: FS operations * fix: default app icons * add back token-read and WSL support * metering fixes * fix: fsEntry * perm scanners and implicators * proper download endpoint * fix: download * fix anti csrft on v2 * fix file extensions, app icons * fold in v1 fixes from origin/main into v2 equivalents Re-applies the v1 fixes that landed on origin/main into their v2 counterparts since the v1 files were deleted on DS/wip during the v2 migration. v1 commits referenced below. - SQLBatcher: flush immediately when queue hits maxBatchSize instead of racing the timer (v1 12f48238). - RedisClient: drop maxRetriesPerRequest from 2 to 1 to shrink failure window (v1 b6776ab4). - ChatCompletionDriver: default minimumCredits to 1 when unset/zero so zero-cost precheck doesn't auto-pass (v1 36bd6073). - OpenAiImageProvider: add gpt-image-2 support — open-ended size rules, token-based cost estimator, arbitrary-size normalizer, isGpt prefix broadened to gpt-image- (v1 f14f1bf4). models.ts auto-merged via rename detection. - AppStore: bump row cache TTL from 5m to 24h (v1 6b3196ed). Not ported: v1 app-object Redis cache (bdfa12b5/b886dde3) — v2's #toClient recomputes filetype_associations/created_from_origin per read; adding a second cache layer is a larger change for a follow-up. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com> * remoe anti-csrf from auth routes that had not used them * more icon fixes * fix worker functionality * fix: app and subdomain es Co-authored-by: Copilot <copilot@github.com> * fix PUT-761 * fix: PUT-748 * fix: rename fsService * Add security back to WorkerDriver * Migrate worker from fsEntry to fs. Fix cache issue * remove ability to create symlinks * strict webdav acl * require auth for wisp * chore: service renames * Add metering back to puter peer api * fix: PUT-760 PUT-749 * fix: PUT-746 * fix: peer cost Co-authored-by: Copilot <copilot@github.com> * fix: 771 * change order of peer controller * fix: create appdata folder for app on get auth token * fix: align delete site and list sites * delete: putility * fix subdomains * Add support for tilde in subdomains, fix subdomain update * cleanup PeerController.ts and fix billing oversight (#2844) * fix: PUT-786 * fix: bugs * fix: issues with multiple subdomain queries, or permission checks * fix: harden response shapes to not contain uneeded fields * fix: move state to redis * fix: missing kv methods + better sec Co-authored-by: Copilot <copilot@github.com> * fix: subdomainStore limit * fix: missing path resolution Co-authored-by: Copilot <copilot@github.com> * fs fixes * fix: undef error * fix fs + cleanup * fix: npm audit fixes * heal path entries where missing Co-authored-by: Copilot <copilot@github.com> * fix: caching Co-authored-by: Copilot <copilot@github.com> * fix: cache inconsistencies Co-authored-by: Copilot <copilot@github.com> * fix: app driver metadata Co-authored-by: Copilot <copilot@github.com> * remove extraneous comma * fix: associated app icons * fix: bad tool call * Add validation to WorkerDriver#getFilePaths * misc fs and auth issues Co-authored-by: Copilot <copilot@github.com> * fix: oidc errors Co-authored-by: Copilot <copilot@github.com> * fix: PUT-797 * fix: legacy appdata_app Co-authored-by: Copilot <copilot@github.com> * fix: add alert logs Co-authored-by: Copilot <copilot@github.com> * fix: error handling * Disable sharecontroller * fix: remove private user identifier for ai * fix: private app fixes * Add backback signup_server * fix: completionId size Co-authored-by: Copilot <copilot@github.com> * fix: revalidate path for oidc * fix: revalidate path for oidc * fix: email validation Co-authored-by: Copilot <copilot@github.com> * fix: user create query * fix: middleware extensions Co-authored-by: Copilot <copilot@github.com> * use x-forwarded-for for req ip forwarded * fix: missing last_activity ts * feat: add cache broadcast to subdomains * fix: update config typing --------- Co-authored-by: Claude Opus 4.6 (1M context) <noreply@anthropic.com> Co-authored-by: ProgrammerIn-wonderland <3838shah@gmail.com> Co-authored-by: Copilot <copilot@github.com> Co-authored-by: Nariman Jelveh <nj@puter.com> Co-authored-by: velzie <velzie@velzie.rip>
2026-09-21 20:55:55 +00:00 · 2026-04-30 12:13:43 -07:00
parent 255dde9e96
commit d4d78ac7db
1076 changed files with 61090 additions and 128544 deletions
@@ -0,0 +1,401 @@
+import { createPool, type Pool } from 'mysql2';
+import { AbstractDatabaseClient, type WriteResult } from './DatabaseClient';
+import { SQLBatcher } from './SQLBatcher.js';
+import type { IConfig } from '../../types';
+
+const RETRIABLE_ERROR_CODES = new Set([
+    'PROTOCOL_CONNECTION_LOST',
+    'PROTOCOL_SEQUENCE_TIMEOUT',
+    'PROTOCOL_ENQUEUE_AFTER_FATAL_ERROR',
+    'ECONNRESET',
+    'ETIMEDOUT',
+    'EPIPE',
+    'ECONNREFUSED',
+    'EHOSTUNREACH',
+    'ENETUNREACH',
+    'EAI_AGAIN',
+]);
+
+const RETRIABLE_ERROR_MESSAGES = [
+    'Connection lost',
+    'read ECONNRESET',
+    'ETIMEDOUT',
+];
+
+type PoolConfig = Parameters<typeof createPool>[0];
+
+enum Configuration {
+    SINGLE,
+    REPLICA,
+}
+
+export class MySQLDatabaseClient extends AbstractDatabaseClient {
+    override readonly engineName = 'mysql';
+
+    private primaryPool!: Pool;
+    private replicaPool!: Pool;
+    private db!: SQLBatcher;
+    private dbReplica!: SQLBatcher;
+    private configuration = Configuration.SINGLE;
+    private shutdownStarted = false;
+    private shutdownTimer: ReturnType<typeof setTimeout> | null = null;
+
+    constructor(config: IConfig) {
+        super(config);
+    }
+
+    // ------------------------------------------------------------------
+    // Lifecycle
+    // ------------------------------------------------------------------
+
+    override async onServerStart(): Promise<void> {
+        const dbConf = this.config.database!;
+
+        this.primaryPool = this.createPool({
+            host: dbConf.host ?? '127.0.0.1',
+            port: dbConf.port ?? 3306,
+            user: dbConf.user ?? 'root',
+            password: dbConf.password ?? '',
+            database: dbConf.database ?? 'puter',
+        });
+        console.log('[mysql] connected to primary');
+
+        this.db = new SQLBatcher(this.primaryPool, 40, 10);
+
+        if (dbConf.replica) {
+            this.replicaPool = this.createPool(dbConf.replica);
+            this.configuration = Configuration.REPLICA;
+            console.log('[mysql] connected to read-replica');
+        } else {
+            this.replicaPool = this.primaryPool;
+            this.configuration = Configuration.SINGLE;
+        }
+
+        this.dbReplica = new SQLBatcher(this.replicaPool, 10, 10);
+    }
+
+    override async onServerPrepareShutdown(): Promise<void> {
+        if (this.shutdownStarted) return;
+        this.shutdownStarted = true;
+
+        // Allow in-flight queries to drain before closing pools
+        const drainMs = 60_000;
+        console.log(
+            `[mysql] draining in-flight queries (${drainMs}ms) before closing pools`,
+        );
+
+        this.shutdownTimer = setTimeout(() => {
+            this.shutdownTimer = null;
+            this.closeCurrentPools('drain').catch((e) =>
+                console.error('[mysql] error closing pools after drain', e),
+            );
+        }, drainMs);
+
+        if (typeof this.shutdownTimer.unref === 'function') {
+            this.shutdownTimer.unref();
+        }
+    }
+
+    override async onServerShutdown(): Promise<void> {
+        if (this.shutdownTimer) {
+            clearTimeout(this.shutdownTimer);
+            this.shutdownTimer = null;
+        }
+        await this.closeCurrentPools('shutdown');
+    }
+
+    // ------------------------------------------------------------------
+    // Query interface
+    // ------------------------------------------------------------------
+
+    override async read(
+        query: string,
+        params: unknown[] = [],
+    ): Promise<Record<string, unknown>[]> {
+        const result = await this.dbReplica.execute(query, params);
+        if (!result) return [];
+        return (result[0] as Record<string, unknown>[]) ?? [];
+    }
+
+    override async pread(
+        query: string,
+        params: unknown[] = [],
+    ): Promise<Record<string, unknown>[]> {
+        const result = await this.db.execute(query, params);
+        if (!result) return [];
+        return (result[0] as Record<string, unknown>[]) ?? [];
+    }
+
+    override async write(
+        query: string,
+        params: unknown[] = [],
+    ): Promise<WriteResult> {
+        const result = await this.db.execute(query, params);
+        const header = result[0] as {
+            insertId?: number;
+            affectedRows?: number;
+        };
+        return {
+            insertId: header.insertId ?? 0,
+            anyRowsAffected: (header.affectedRows ?? 0) > 0,
+        };
+    }
+
+    override async batchWrite(
+        entries: { statement: string; values: unknown[] }[],
+    ): Promise<void> {
+        if (entries.length === 0) return;
+        // Bypass the SQLBatcher: it coalesces queries from unrelated callers
+        // into a single multi-statement string, which is incompatible with
+        // wrapping a transaction around just *our* statements. Acquire a
+        // dedicated connection so BEGIN/COMMIT/ROLLBACK only scope `entries`.
+        const conn = await this.primaryPool.promise().getConnection();
+        try {
+            await conn.beginTransaction();
+            try {
+                for (const { statement, values } of entries) {
+                    await conn.execute(statement, values);
+                }
+                await conn.commit();
+            } catch (err) {
+                await conn.rollback().catch(() => {});
+                throw err;
+            }
+        } finally {
+            conn.release();
+        }
+    }
+
+    override async tryHardRead(
+        query: string,
+        params: unknown[] = [],
+    ): Promise<Record<string, unknown>[]> {
+        if (this.configuration === Configuration.SINGLE) {
+            return this.read(query, params);
+        }
+
+        // Run both reads in parallel — prefer replica when it returns rows,
+        // otherwise fall back to primary to handle replication lag.
+        const primaryPromise = this.db.execute(query, params);
+        try {
+            const replicaResult = await this.dbReplica.execute(query, params);
+            if (
+                Array.isArray(replicaResult?.[0]) &&
+                (replicaResult[0] as unknown[]).length > 0
+            ) {
+                primaryPromise.catch(() => {}); // suppress unhandled rejection
+                return replicaResult[0] as Record<string, unknown>[];
+            }
+        } catch {
+            // fall through to primary
+        }
+
+        const primaryResult = await primaryPromise;
+        return (primaryResult?.[0] as Record<string, unknown>[]) ?? [];
+    }
+
+    // ------------------------------------------------------------------
+    // Pool management
+    // ------------------------------------------------------------------
+
+    private createPool(poolConfig: PoolConfig): Pool {
+        return createPool({
+            maxPreparedStatements: 900,
+            ...poolConfig,
+            multipleStatements: true,
+        } as PoolConfig);
+    }
+
+    /** Reinitialize the primary pool (e.g. after a health-check failure). */
+    reinitPrimary(): void {
+        if (this.shutdownStarted) return;
+
+        const dbConf = this.config.database!;
+        const previous = this.primaryPool;
+        this.primaryPool = this.createPool({
+            host: dbConf.host ?? '127.0.0.1',
+            port: dbConf.port ?? 3306,
+            user: dbConf.user ?? 'root',
+            password: dbConf.password ?? '',
+            database: dbConf.database ?? 'puter',
+        });
+        this.db = new SQLBatcher(this.primaryPool, 40, 10);
+
+        if (this.configuration === Configuration.SINGLE) {
+            this.replicaPool = this.primaryPool;
+            this.dbReplica = new SQLBatcher(this.primaryPool, 10, 10);
+        }
+
+        if (previous && previous !== this.primaryPool) {
+            this.closePool(previous, 'reinit:primary').catch(() => {});
+        }
+    }
+
+    /** Reinitialize the replica pool. */
+    reinitReplica(): void {
+        if (this.shutdownStarted || !this.config.database?.replica) return;
+
+        const previous = this.replicaPool;
+        this.replicaPool = this.createPool(this.config.database.replica);
+        this.dbReplica = new SQLBatcher(this.replicaPool, 10);
+
+        if (
+            previous &&
+            previous !== this.replicaPool &&
+            previous !== this.primaryPool
+        ) {
+            this.closePool(previous, 'reinit:replica').catch(() => {});
+        }
+    }
+
+    // ------------------------------------------------------------------
+    // Retry helpers (for health checks or resilient reads)
+    // ------------------------------------------------------------------
+
+    static isRetriableError(error: unknown): boolean {
+        const code = (error as { code?: string })?.code;
+        if (code && RETRIABLE_ERROR_CODES.has(code)) return true;
+
+        const msg = String((error as Error)?.message ?? '');
+        return RETRIABLE_ERROR_MESSAGES.some((m) => msg.includes(m));
+    }
+
+    async readWithRetry(
+        label: string,
+        operation: () => Promise<unknown[]>,
+        opts?: {
+            maxAttempts?: number;
+            baseBackoffMs?: number;
+            maxBackoffMs?: number;
+            jitterRatio?: number;
+        },
+    ): Promise<unknown[]> {
+        const maxAttempts = opts?.maxAttempts ?? 3;
+        const baseBackoffMs = opts?.baseBackoffMs ?? 100;
+        const maxBackoffMs = opts?.maxBackoffMs ?? 500;
+        const jitterRatio = opts?.jitterRatio ?? 0.2;
+
+        let attempt = 1;
+
+        while (true) {
+            try {
+                return await operation();
+            } catch (error) {
+                if (this.shutdownStarted) throw error;
+                if (
+                    attempt >= maxAttempts ||
+                    !MySQLDatabaseClient.isRetriableError(error)
+                )
+                    throw error;
+
+                const raw = baseBackoffMs * 2 ** (attempt - 1);
+                const capped = Math.min(maxBackoffMs, raw);
+                const window = Math.round(capped * jitterRatio);
+                const jitter =
+                    window === 0
+                        ? 0
+                        : Math.floor(Math.random() * (window * 2 + 1)) - window;
+                const delay = Math.max(0, capped + jitter);
+
+                console.warn(
+                    `[${label}] transient mysql error (${(error as { code?: string })?.code ?? 'unknown'}); retry ${attempt + 1}/${maxAttempts} in ${delay}ms`,
+                );
+                await new Promise((r) => setTimeout(r, delay));
+                attempt++;
+            }
+        }
+    }
+
+    // ------------------------------------------------------------------
+    // Internal pool lifecycle
+    // ------------------------------------------------------------------
+
+    private async closePool(
+        pool: Pool,
+        label: string,
+        timeoutMs: number | null = null,
+    ): Promise<void> {
+        if (!pool) return;
+
+        await new Promise<void>((resolve, reject) => {
+            let settled = false;
+            let timer: ReturnType<typeof setTimeout> | null = null;
+
+            const finish = (err?: unknown) => {
+                if (settled) return;
+                settled = true;
+                if (timer) clearTimeout(timer);
+                if (err) reject(err);
+                else resolve();
+            };
+
+            if (timeoutMs !== null) {
+                timer = setTimeout(() => {
+                    console.warn(
+                        `[mysql] timed out closing pool (${label}); forcing`,
+                    );
+                    this.forceDestroyConnections(pool, `${label}:timeout`);
+                    finish();
+                }, timeoutMs);
+            }
+
+            try {
+                pool.end((err) => finish(err));
+            } catch (err) {
+                finish(err);
+            }
+        });
+    }
+
+    private forceDestroyConnections(pool: Pool, label: string): void {
+        // mysql2 internal — _allConnections is a CircularBuffer
+        const all = (
+            pool as unknown as {
+                _allConnections?: {
+                    forEach: (fn: (c: { destroy: () => void }) => void) => void;
+                };
+            }
+        )._allConnections;
+        if (!all || typeof all.forEach !== 'function') return;
+
+        let count = 0;
+        all.forEach((conn) => {
+            try {
+                conn.destroy();
+                count++;
+            } catch {
+                // no-op
+            }
+        });
+        if (count > 0)
+            console.warn(
+                `[mysql] force-closed ${count} connections (${label})`,
+            );
+    }
+
+    private async closeCurrentPools(reason: string): Promise<void> {
+        const timeoutMs = reason.startsWith('signal:') ? 45_000 : null;
+        const tasks: Promise<void>[] = [];
+
+        if (this.primaryPool) {
+            tasks.push(
+                this.closePool(
+                    this.primaryPool,
+                    `${reason}:primary`,
+                    timeoutMs,
+                ),
+            );
+        }
+        if (this.replicaPool && this.replicaPool !== this.primaryPool) {
+            tasks.push(
+                this.closePool(
+                    this.replicaPool,
+                    `${reason}:replica`,
+                    timeoutMs,
+                ),
+            );
+        }
+
+        await Promise.all(tasks);
+    }
+}