From 5cc887c97ceef5d3fbd953e130e62ec6b161e8d1 Mon Sep 17 00:00:00 2001 From: Briant Diehl Date: Thu, 4 Jun 2026 13:39:06 -0600 Subject: [PATCH] refactor(packages): factory + per-package env schema for base packages Convert the civitai-* base packages from eager singletons into app-agnostic factories with their own zod env schemas, and wire them into the main app via thin shims. Packages now import only external deps (+ @civitai/db-schema); the app injects behavior (loggers, Flipt resolver, slow-query sink) and owns HMR globals + the Next build guard. - Phase 0: pnpm-workspace.yaml, per-package package.json + index barrels, @civitai/* tsconfig paths, transpilePackages - axiom/redis/db/clickhouse: createX() factories reading package-owned env.ts (z.prettifyError); Partial overrides - db: createPrismaClients + config-driven getClient pool factory; kv-helpers.ts breaks the client<->db-helpers cycle; limitConcurrency vendored; pg singletons -> app shims - clickhouse: base client stays; Tracker (auth/session/schema-coupled) extracted to src/server/clickhouse/tracker.ts - telemetry: prom helpers stay; DB pool-gauge block -> src/server/prom/client.ts - prisma generate paths updated for the db-schema package; generated slim schema gitignored Co-Authored-By: Claude Opus 4.8 (1M context) --- .gitignore | 1 + next.config.mjs | 9 +- package.json | 3 +- packages/civitai-axiom/package.json | 7 + packages/civitai-axiom/src/client.ts | 67 +- packages/civitai-axiom/src/env.ts | 35 + packages/civitai-axiom/src/index.ts | 1 + packages/civitai-clickhouse/package.json | 7 + packages/civitai-clickhouse/src/client.ts | 745 ++---------------- packages/civitai-clickhouse/src/env.ts | 28 + packages/civitai-clickhouse/src/index.ts | 1 + packages/civitai-db-schema/package.json | 7 + .../prisma/schema.full.prisma | 4 +- packages/civitai-db-schema/src/index.ts | 6 + packages/civitai-db/package.json | 10 + packages/civitai-db/src/client.ts | 152 ++-- .../civitai-db/src/concurrency-helpers.ts | 68 ++ packages/civitai-db/src/db-helpers.ts | 125 ++- packages/civitai-db/src/env.ts | 63 ++ packages/civitai-db/src/index.ts | 4 + packages/civitai-db/src/kv-helpers.ts | 49 ++ packages/civitai-redis/package.json | 7 + packages/civitai-redis/src/client.ts | 102 +-- packages/civitai-redis/src/env.ts | 37 + packages/civitai-redis/src/index.ts | 1 + packages/civitai-telemetry/package.json | 7 + packages/civitai-telemetry/src/client.ts | 102 +-- packages/civitai-telemetry/src/index.ts | 2 + pnpm-lock.yaml | 16 + pnpm-workspace.yaml | 4 + scripts/generate-slim-schema.js | 10 +- src/server/clickhouse/client.ts | 27 + src/server/clickhouse/tracker.ts | 641 +++++++++++++++ src/server/db/client.ts | 28 + .../src => src/server/db}/datapacketDb.ts | 17 +- src/server/db/db-helpers.ts | 15 + .../src => src/server/db}/notifDb.ts | 22 +- .../civitai-db/src => src/server/db}/pgDb.ts | 27 +- src/server/logging/client.ts | 12 + src/server/prom/client.ts | 106 +++ src/server/redis/client.ts | 36 + src/server/utils/otel-helpers.ts | 3 + src/shared/utils/prisma/enums.ts | 3 + src/shared/utils/prisma/models.ts | 3 + tsconfig.json | 15 +- 45 files changed, 1562 insertions(+), 1073 deletions(-) create mode 100644 packages/civitai-axiom/package.json create mode 100644 packages/civitai-axiom/src/env.ts create mode 100644 packages/civitai-axiom/src/index.ts create mode 100644 packages/civitai-clickhouse/package.json create mode 100644 packages/civitai-clickhouse/src/env.ts create mode 100644 packages/civitai-clickhouse/src/index.ts create mode 100644 packages/civitai-db-schema/package.json create mode 100644 packages/civitai-db-schema/src/index.ts create mode 100644 packages/civitai-db/package.json create mode 100644 packages/civitai-db/src/concurrency-helpers.ts create mode 100644 packages/civitai-db/src/env.ts create mode 100644 packages/civitai-db/src/index.ts create mode 100644 packages/civitai-db/src/kv-helpers.ts create mode 100644 packages/civitai-redis/package.json create mode 100644 packages/civitai-redis/src/env.ts create mode 100644 packages/civitai-redis/src/index.ts create mode 100644 packages/civitai-telemetry/package.json create mode 100644 packages/civitai-telemetry/src/index.ts create mode 100644 pnpm-workspace.yaml create mode 100644 src/server/clickhouse/client.ts create mode 100644 src/server/clickhouse/tracker.ts create mode 100644 src/server/db/client.ts rename {packages/civitai-db/src => src/server/db}/datapacketDb.ts (51%) create mode 100644 src/server/db/db-helpers.ts rename {packages/civitai-db/src => src/server/db}/notifDb.ts (58%) rename {packages/civitai-db/src => src/server/db}/pgDb.ts (64%) create mode 100644 src/server/logging/client.ts create mode 100644 src/server/prom/client.ts create mode 100644 src/server/redis/client.ts create mode 100644 src/server/utils/otel-helpers.ts create mode 100644 src/shared/utils/prisma/enums.ts create mode 100644 src/shared/utils/prisma/models.ts diff --git a/.gitignore b/.gitignore index cba49cd3b0..1f013fbe01 100644 --- a/.gitignore +++ b/.gitignore @@ -14,6 +14,7 @@ node_modules # Generated slim schema - edit schema.full.prisma instead prisma/schema.prisma +packages/civitai-db-schema/prisma/schema.prisma # next.js /.next/ diff --git a/next.config.mjs b/next.config.mjs index bb4c3a1d51..06054a8b0b 100644 --- a/next.config.mjs +++ b/next.config.mjs @@ -103,7 +103,14 @@ export default defineNextConfig( // removeConsole: true, } : {}, - transpilePackages: [], + transpilePackages: [ + '@civitai/db-schema', + '@civitai/db', + '@civitai/redis', + '@civitai/clickhouse', + '@civitai/axiom', + '@civitai/telemetry', + ], experimental: { // scrollRestoration: true, cpus: 8, diff --git a/package.json b/package.json index b10765e2d9..3056e55607 100644 --- a/package.json +++ b/package.json @@ -66,7 +66,8 @@ "ts-script": "NODE_ENV=development tsx" }, "prisma": { - "seed": "ts-node --compiler-options {\"module\":\"CommonJS\"} prisma/seed.ts" + "schema": "packages/civitai-db-schema/prisma/schema.prisma", + "seed": "ts-node --compiler-options {\"module\":\"CommonJS\"} packages/civitai-db-schema/prisma/seed.ts" }, "lint-staged": { "**/*.{ts,tsx}": "tsc-files --noEmit" diff --git a/packages/civitai-axiom/package.json b/packages/civitai-axiom/package.json new file mode 100644 index 0000000000..b12d849df7 --- /dev/null +++ b/packages/civitai-axiom/package.json @@ -0,0 +1,7 @@ +{ + "name": "@civitai/axiom", + "version": "0.0.0", + "private": true, + "main": "./src/index.ts", + "types": "./src/index.ts" +} diff --git a/packages/civitai-axiom/src/client.ts b/packages/civitai-axiom/src/client.ts index 6b9bfa0573..0ef618ff15 100644 --- a/packages/civitai-axiom/src/client.ts +++ b/packages/civitai-axiom/src/client.ts @@ -1,14 +1,5 @@ import { Client } from '@axiomhq/axiom-node'; -import { isProd } from '~/env/other'; -import { env } from '~/env/server'; - -const shouldConnect = !env.IS_BUILD && env.AXIOM_TOKEN && env.AXIOM_ORG_ID; -const axiom = shouldConnect - ? new Client({ - token: env.AXIOM_TOKEN, - orgId: env.AXIOM_ORG_ID, - }) - : null; +import { axiomEnv, type AxiomConfig } from './env'; /** * Extract only safe primitive fields from an error for logging. @@ -35,24 +26,46 @@ export function safeError(e: unknown): MixedObject | undefined { return { message: String(e) }; } -export async function logToAxiom(data: MixedObject, datastream?: string) { - const sendData = { pod: env.PODNAME, ...data }; - if (isProd) { - if (!axiom) return; - datastream ??= env.AXIOM_DATASTREAM; - if (!datastream) return; +export type AxiomLogger = { + logToAxiom: (data: MixedObject, datastream?: string) => Promise; + safeError: typeof safeError; +}; - // Write stderr BEFORE awaiting Axiom — when Axiom is degraded, - // ingestEvents rejects and the rest of this function never runs. - // Loki ingest depends on the stderr line; without this ordering, - // the Grafana alerts that consume `{ "name": "sysredis-fail-open", - // ... }` go silent during the exact multi-service incident class - // they exist to handle (sysRedis + Axiom both down). - if (process.env.LOG_ERRORS_TO_STDOUT === 'true') - console.error(JSON.stringify({ _axiom: datastream, ...sendData })); +/** + * Build an Axiom logger. Config defaults come from the package's own env schema + * (./env); pass a `Partial` to override any value per call (tests, + * multi-instance, alternate config sources). Axiom has no injected app-behavior + * deps (it *is* the logger). See the `~/server/logging/client` shim. + */ +export function createAxiomLogger(overrides: Partial = {}): AxiomLogger { + const config = { ...axiomEnv, ...overrides }; - await axiom.ingestEvents(datastream, sendData); - } else { - console.log('logToAxiom', sendData); + const axiom = + config.token && config.orgId + ? new Client({ token: config.token, orgId: config.orgId }) + : null; + + async function logToAxiom(data: MixedObject, datastream?: string) { + const sendData = { pod: config.podName, ...data }; + if (config.isProd) { + if (!axiom) return; + datastream ??= config.datastream; + if (!datastream) return; + + // Write stderr BEFORE awaiting Axiom — when Axiom is degraded, + // ingestEvents rejects and the rest of this function never runs. + // Loki ingest depends on the stderr line; without this ordering, + // the Grafana alerts that consume `{ "name": "sysredis-fail-open", + // ... }` go silent during the exact multi-service incident class + // they exist to handle (sysRedis + Axiom both down). + if (config.logErrorsToStdout) + console.error(JSON.stringify({ _axiom: datastream, ...sendData })); + + await axiom.ingestEvents(datastream, sendData); + } else { + console.log('logToAxiom', sendData); + } } + + return { logToAxiom, safeError }; } diff --git a/packages/civitai-axiom/src/env.ts b/packages/civitai-axiom/src/env.ts new file mode 100644 index 0000000000..bd867d7f25 --- /dev/null +++ b/packages/civitai-axiom/src/env.ts @@ -0,0 +1,35 @@ +// Package-owned env schema. Any app that uses @civitai/axiom validates these +// vars the same way, so every app logs to Axiom with identical config. +import * as z from 'zod'; + +const booleanString = z.preprocess((val) => val === true || val === 'true', z.boolean()); + +// Every env var the Axiom logger reads is declared here so it's validated on +// deployment. App *behavior* (loggers, policy callbacks) would be injected at the +// factory instead — but those are functions, not env values, and this package has none. +const schema = z.object({ + AXIOM_TOKEN: z.string().optional(), + AXIOM_ORG_ID: z.string().optional(), + AXIOM_DATASTREAM: z.string().optional(), + PODNAME: z.string().optional(), + LOG_ERRORS_TO_STDOUT: booleanString.default(false), +}); + +const parsed = schema.safeParse(process.env); +if (!parsed.success) { + throw new Error('[@civitai/axiom] Invalid environment variables:\n' + z.prettifyError(parsed.error)); +} + +// Normalized, env-derived defaults. The factory accepts a Partial to +// override any of these per call (tests, multi-instance, alternate config sources). +export const axiomEnv = { + token: parsed.data.AXIOM_TOKEN, + orgId: parsed.data.AXIOM_ORG_ID, + datastream: parsed.data.AXIOM_DATASTREAM, + podName: parsed.data.PODNAME, + logErrorsToStdout: parsed.data.LOG_ERRORS_TO_STDOUT, + // NODE_ENV is a universal Node convention (not Next-specific), so it's fine for a package. + isProd: process.env.NODE_ENV === 'production', +}; + +export type AxiomConfig = typeof axiomEnv; diff --git a/packages/civitai-axiom/src/index.ts b/packages/civitai-axiom/src/index.ts new file mode 100644 index 0000000000..4f1cce44fa --- /dev/null +++ b/packages/civitai-axiom/src/index.ts @@ -0,0 +1 @@ +export * from './client'; diff --git a/packages/civitai-clickhouse/package.json b/packages/civitai-clickhouse/package.json new file mode 100644 index 0000000000..631c889009 --- /dev/null +++ b/packages/civitai-clickhouse/package.json @@ -0,0 +1,7 @@ +{ + "name": "@civitai/clickhouse", + "version": "0.0.0", + "private": true, + "main": "./src/index.ts", + "types": "./src/index.ts" +} diff --git a/packages/civitai-clickhouse/src/client.ts b/packages/civitai-clickhouse/src/client.ts index e0264abec4..e70d3a937d 100644 --- a/packages/civitai-clickhouse/src/client.ts +++ b/packages/civitai-clickhouse/src/client.ts @@ -1,31 +1,9 @@ +// Base ClickHouse client. The Tracker (request/session/schema-coupled event recorder) +// lives in the app (src/server/clickhouse/tracker.ts), not here. import type { ClickHouseClient } from '@clickhouse/client'; import { createClient } from '@clickhouse/client'; -import type { NextApiRequest, NextApiResponse } from 'next'; -import type { Session } from 'next-auth'; -import requestIp from 'request-ip'; -import { isProd } from '~/env/other'; -import { env } from '~/env/server'; -import type { NewOrderImageRatingStatus, NsfwLevel } from '~/server/common/enums'; -import type { AllModKeys } from '~/server/jobs/entity-moderation'; -import { logToAxiom } from '~/server/logging/client'; -import { sleep } from '~/utils/errorHandling'; -import type { AddImageRatingInput } from '~/server/schema/games/new-order.schema'; -import type { ProhibitedSources } from '~/server/schema/user.schema'; -import type { NsfwLevelDeprecated } from '~/shared/constants/browsingLevel.constants'; -import dayjs from '~/shared/utils/dayjs'; -import type { - ArticleEngagementType, - BountyEngagementType, - EntityMetric_EntityType_Type, - EntityMetric_MetricType_Type, - EntityType, - NewOrderRankType, - ReportReason, - ReportStatus, - ReviewReactions, -} from '~/shared/utils/prisma/enums'; -import { createLogger } from '~/utils/logging'; -import { getServerAuthSession } from '~/server/auth/get-server-auth-session'; +import dayjs from 'dayjs'; +import { clickhouseEnv, type ClickhouseConfig } from './env'; export type CustomClickHouseClient = ClickHouseClient & { $query: ( @@ -35,73 +13,12 @@ export type CustomClickHouseClient = ClickHouseClient & { $exec: (query: TemplateStringsArray | string, ...values: any[]) => Promise; }; -declare global { - // eslint-disable-next-line no-var, vars-on-top - var globalClickhouse: CustomClickHouseClient | undefined; -} +export type ClickhouseLogFn = (message: string, ...args: unknown[]) => void; -const log = createLogger('clickhouse', 'blue'); - -function getClickHouse() { - console.log('Creating ClickHouse client'); - const client = createClient({ - host: env.CLICKHOUSE_HOST, - username: env.CLICKHOUSE_USERNAME, - password: env.CLICKHOUSE_PASSWORD, - clickhouse_settings: { - async_insert: 1, - wait_for_async_insert: 0, - output_format_json_quote_64bit_integers: 0, // otherwise they come as strings - }, - }) as CustomClickHouseClient; - - client.$query = async function ( - query: TemplateStringsArray | string, - ...values: any[] - ) { - if (typeof query !== 'string') { - query = query.reduce((acc, part, i) => acc + part + formatSqlType(values[i] ?? ''), ''); - } - - log('$query', query); - - try { - const response = await client.query({ - query, - format: 'JSONEachRow', - }); - const data = await response?.json(); - return data; - } catch (e) { - const error = e as Error; - throw new Error(`ClickHouse query failed: ${error.message}\nQuery: ${query}`); - } - }; - - client.$exec = async function (query: TemplateStringsArray | string, ...values: any[]) { - if (typeof query !== 'string') { - query = query.reduce((acc, part, i) => acc + part + formatSqlType(values[i] ?? ''), ''); - } - - log('$exec', query); - - await client.exec({ - query, - }); - }; - - return client; -} - -export let clickhouse: CustomClickHouseClient | undefined; -const shouldConnect = !env.IS_BUILD && env.CLICKHOUSE_HOST && env.CLICKHOUSE_USERNAME; -if (shouldConnect) { - if (isProd) clickhouse = getClickHouse(); - else { - if (!global.globalClickhouse) global.globalClickhouse = getClickHouse(); - clickhouse = global.globalClickhouse; - } -} +export type CreateClickhouseClientOptions = Partial & { + /** Debug logger (app-defined). Defaults to a no-op. */ + log?: ClickhouseLogFn; +}; function formatSqlType(value: any): string { // Catch any dates being passed in as a string @@ -121,613 +38,59 @@ function formatSqlType(value: any): string { return value; } -export type ViewType = - | 'ProfileView' - | 'ImageView' - | 'PostView' - | 'ModelView' - | 'ModelVersionView' - | 'ArticleView' - | 'CollectionView' - | 'BountyView' - | 'BountyEntryView'; +/** + * Build the base ClickHouse client. Connection config defaults come from the package + * env schema (./env, overridable via options); the debug logger is injected. HMR/global + * caching and the Next build guard live in the app shim. See `~/server/clickhouse/client`. + */ +export function createClickhouseClient( + options: CreateClickhouseClientOptions = {} +): CustomClickHouseClient { + const { log: logOption, ...envOverrides } = options; + const config = { ...clickhouseEnv, ...envOverrides }; + const log: ClickhouseLogFn = logOption ?? (() => {}); -export type UserActivityType = - | 'Registration' - | 'Login' - | 'Account closure' - | 'Subscribe' - | 'Cancel' - | 'Donate' - | 'Adjust Moderated Content Settings' - | 'Banned' - | 'Unbanned' - | 'Muted' - | 'Unmuted' - | 'RemoveContent' - | 'ExcludedFromLeaderboard' - | 'UnexcludedFromLeaderboard'; -export type ModelVersionActivty = 'Create' | 'Publish' | 'Download' | 'Unpublish' | 'HideDownload'; -export type ModelActivty = - | 'Create' - | 'Publish' - | 'Update' - | 'Unpublish' - | 'Archive' - | 'Takedown' - | 'Delete' - | 'PermanentDelete' - | 'Transfer'; -export type ResourceReviewType = 'Create' | 'Delete' | 'Exclude' | 'Include' | 'Update'; -export type ReactionType = - | 'Images_Create' - | 'Images_Delete' - | 'Comment_Create' - | 'Comment_Delete' - | 'Review_Create' - | 'Review_Delete' - | 'Question_Create' - | 'Question_Delete' - | 'Answer_Create' - | 'Answer_Delete' - | 'BountyEntry_Create' - | 'BountyEntry_Delete' - | 'Article_Create' - | 'Article_Delete'; -export type ReportType = 'Create' | 'StatusChange'; -export type ModelEngagementType = 'Hide' | 'Favorite' | 'Delete' | 'Notify'; -export type TagEngagementType = 'Hide' | 'Allow'; -export type UserEngagementType = 'Follow' | 'Hide' | 'Delete'; -export type CommentType = - | 'Model' - | 'Image' - | 'Post' - | 'Comment' - | 'Review' - | 'Bounty' - | 'BountyEntry'; -export type CommentActivity = 'Create' | 'Delete' | 'Update' | 'Hide' | 'Unhide'; -export type PostActivityType = 'Create' | 'Publish' | 'Tags' | 'Delete'; -export type ImageActivityType = - | 'Create' - | 'Delete' - | 'DeleteTOS' - | 'Tags' - | 'Resources' - | 'Restore'; -export type QuestionType = 'Create' | 'Delete'; -export type AnswerType = 'Create' | 'Delete'; -export type PartnerActivity = 'Run' | 'Update'; -export type BountyActivity = 'Create' | 'Update' | 'Delete' | 'Expire' | 'Refund'; -export type BountyEntryActivity = 'Create' | 'Update' | 'Delete' | 'Award'; -export type BountyBenefactorActivity = 'Create'; + console.log('Creating ClickHouse client'); + const client = createClient({ + host: config.host, + username: config.username, + password: config.password, + clickhouse_settings: { + async_insert: 1, + wait_for_async_insert: 0, + output_format_json_quote_64bit_integers: 0, // otherwise they come as strings + }, + }) as CustomClickHouseClient; -export type FileActivity = 'Download'; -export type ModelFileActivity = 'Create' | 'Delete' | 'Update'; + client.$query = async function ( + query: TemplateStringsArray | string, + ...values: any[] + ) { + if (typeof query !== 'string') { + query = query.reduce((acc, part, i) => acc + part + formatSqlType(values[i] ?? ''), ''); + } -export const ActionType = [ - 'AddToBounty_Click', - 'AddToBounty_Confirm', - 'AwardBounty_Click', - 'AwardBounty_Confirm', - 'Tip_Click', - 'Tip_Confirm', - 'TipInteractive_Click', - 'TipInteractive_Cancel', - 'NotEnoughFunds', - 'PurchaseFunds_Cancel', - 'PurchaseFunds_Confirm', - 'LoginRedirect', - 'Membership_Cancel', - 'Membership_Downgrade', - 'CSAM_Help_Triggered', - 'ProfanitySearch', - 'BuzzLimit_Set', - // Generation funnel telemetry — top-of-funnel clicks + form submission. - // Joined to orchestration.jobs / images_created downstream by userId + ts. - 'Model_Create_Click', - 'Image_Remix_Click', - 'Generator_Submit', -] as const; -export type ActionType = (typeof ActionType)[number]; + log('$query', query); -export type TrackRequest = { - userId: number; - ip: string; - userAgent: string; -}; - -/** Track a webhook event to ClickHouse (fire and forget) */ -export async function trackWebhookEvent(type: string, payload: string) { - if (!clickhouse) return; - - try { - await clickhouse.insert({ - table: 'webhook_events_buffer', - values: [{ type, payload }], - format: 'JSONEachRow', - }); - } catch (error: any) { - console.error(`Failed to track ${type} webhook to ClickHouse:`, error.message); - } -} - -export class Tracker { - private actor: TrackRequest = { - userId: 0, - ip: 'unknown', - userAgent: 'unknown', + try { + const response = await client.query({ query, format: 'JSONEachRow' }); + const data = await response?.json(); + return data; + } catch (e) { + const error = e as Error; + throw new Error(`ClickHouse query failed: ${error.message}\nQuery: ${query}`); + } }; - private session: Session | null = null; - private req: NextApiRequest | undefined; - private res: NextApiResponse | undefined; - private async resolveSession() { - if (!this.session && this.req && this.res) { - try { - await getServerAuthSession({ req: this.req, res: this.res }).then((session) => { - this.session = session; - this.actor.userId = session?.user?.id ?? this.actor.userId; - return session; - }); - } catch (e) { - const error = e as Error; - logToAxiom( - { - type: 'error', - name: 'Failed session', - message: error.message, - stack: error.stack, - cause: error.cause, - }, - 'clickhouse' - ); - } + client.$exec = async function (query: TemplateStringsArray | string, ...values: any[]) { + if (typeof query !== 'string') { + query = query.reduce((acc, part, i) => acc + part + formatSqlType(values[i] ?? ''), ''); } - } - constructor(req?: NextApiRequest, res?: NextApiResponse) { - if (req && res) { - this.req = req; - this.res = res; - this.actor.ip = requestIp.getClientIp(req) ?? this.actor.ip; - this.actor.userAgent = req.headers['user-agent'] ?? this.actor.userAgent; - } - } + log('$exec', query); - private async send( - table: string, - data: object | ((args: { session: Session | null; actor: TrackRequest }) => object) - ) { - if (!env.CLICKHOUSE_TRACKER_URL) return; - await this.resolveSession(); + await client.exec({ query }); + }; - const body = - typeof data === 'function' ? data({ session: this.session, actor: this.actor }) : data; - const url = `${env.CLICKHOUSE_TRACKER_URL}/track/${table}`; - - // Fire-and-forget at the call site, but the inner attempt loop checks - // HTTP status (not just network errors) and retries 5xx with backoff. - // Prior version only handled network rejection from fetch(), so any - // 5xx response from the tracker — common when NATS publish ack times - // out — was silently dropped. - void this.sendWithRetry(url, body, table); - } - - private async sendWithRetry( - url: string, - body: object, - table: string, - attempt = 1 - ): Promise { - const MAX_ATTEMPTS = 3; - const baseDelayMs = 250; - - try { - const res = await fetch(url, { - method: 'POST', - body: JSON.stringify(body), - headers: { 'Content-Type': 'application/json' }, - }); - - if (res.ok) return; - - // 4xx: tracker rejected the payload. Retrying won't help. Log and bail. - if (res.status >= 400 && res.status < 500) { - const errBody = await res.text().catch(() => ''); - logToAxiom( - { - type: 'warning', - name: 'Failed to track (4xx)', - details: { table, status: res.status, attempt, response: errBody.slice(0, 500) }, - message: `Tracker returned ${res.status}`, - }, - 'clickhouse' - ).catch(() => {}); - return; - } - - // 5xx: transient — NATS publish timeout, JetStream rejection, etc. - if (attempt < MAX_ATTEMPTS) { - await sleep(baseDelayMs * 2 ** (attempt - 1) + Math.random() * baseDelayMs); - return this.sendWithRetry(url, body, table, attempt + 1); - } - - const errBody = await res.text().catch(() => ''); - logToAxiom( - { - type: 'error', - name: 'Failed to track (5xx, exhausted)', - details: { table, status: res.status, attempts: attempt, response: errBody.slice(0, 500) }, - message: `Tracker returned ${res.status} after ${attempt} attempts`, - }, - 'clickhouse' - ).catch(() => {}); - } catch (e) { - const error = e as Error; - // Network-level failure. Retry the same as 5xx. - if (attempt < MAX_ATTEMPTS) { - await sleep(baseDelayMs * 2 ** (attempt - 1) + Math.random() * baseDelayMs); - return this.sendWithRetry(url, body, table, attempt + 1); - } - logToAxiom( - { - type: 'error', - name: 'Failed to track (network, exhausted)', - details: { table, attempts: attempt }, - message: error.message, - stack: error.stack, - cause: error.cause, - }, - 'clickhouse' - ).catch(() => {}); - } - } - - private async sendMany( - table: string, - data: object[] | ((args: { session: Session | null; actor: TrackRequest }) => object[]) - ) { - if (!clickhouse) return; - await this.resolveSession(); - const values = - typeof data === 'function' ? data({ session: this.session, actor: this.actor }) : data; - - try { - await clickhouse.insert({ - table, - values, - format: 'JSONEachRow', - }); - } catch (e) { - const error = e as Error; - logToAxiom( - { - type: 'error', - name: 'Failed to track', - details: { table, data: JSON.stringify(data) }, - message: error.message, - stack: error.stack, - cause: error.cause, - }, - 'clickhouse' - ).catch(); - } - } - - private async track( - table: string, - custom: object | ((session: Session | null) => object), - options?: { skipActorMeta: boolean } - ): Promise { - const { skipActorMeta = false } = options ?? {}; - - await this.send(table, ({ session, actor }) => { - const actorMeta = skipActorMeta ? { userId: actor.userId } : { ...actor }; - const customData = typeof custom === 'function' ? custom(session) : custom; - - return { - ...actorMeta, - ...customData, - }; - }); - } - - private async trackMany( - table: string, - custom: object[] | ((session: Session | null) => object[]), - options?: { skipActorMeta: boolean } - ) { - const { skipActorMeta = false } = options ?? {}; - - await this.sendMany(table, ({ session, actor }) => { - const actorMeta = skipActorMeta ? { userId: actor.userId } : { ...actor }; - const customData = typeof custom === 'function' ? custom(session) : custom; - return customData.map((custom) => ({ - ...actorMeta, - ...custom, - })); - }); - } - - public view(values: { type: ViewType; entityType: EntityType; entityId: number }) { - return this.track('views', values); - } - - public pageView(values: { - pageId: string; - path: string; - host: string; - ads: boolean; - country: string; - duration: number; - windowWidth: number; - windowHeight: number; - }) { - return this.send('pageViews', ({ session, actor }) => { - return { - userId: actor.userId, - memberType: session?.user?.tier ?? 'undefined', - ip: actor.ip, - ...values, - }; - }); - } - - public action(values: { type: ActionType; details?: any }) { - const { details, ...rest } = values; - return this.track('actions', { - ...rest, - details: - details != null ? (typeof details === 'string' ? details : JSON.stringify(details)) : '', - }); - } - - public activity(activity: string) { - return this.track('activities', { activity }); - } - - public bugReport(values: { bugId: number; status: string }) { - return this.track('bugReports', values); - } - - public modelEvent(values: { type: ModelActivty; modelId: number; nsfw: boolean }) { - return this.track('modelEvents', values); - } - - public redeemableCode(activity: string, details: { quantity?: number; code?: string }) { - return this.track('redeemableCodes', { activity, ...details }); - } - - public modelVersionEvent(values: { - type: ModelVersionActivty; - modelId: number; - modelVersionId: number; - nsfw: boolean; - earlyAccess?: boolean; - time?: Date; - fileId?: number; - }) { - return this.track('modelVersionEvents', values); - } - - public partnerEvent(values: { - type: PartnerActivity; - partnerId: number; - modelId?: number; - modelVersionId?: number; - nsfw?: boolean; - }) { - return this.track('partnerEvents', values); - } - - public userActivity(values: { - type: UserActivityType; - targetUserId: number; - source?: string; - landingPage?: string; - }) { - return this.track('userActivities', values); - } - - public resourceReview(values: { - type: ResourceReviewType; - modelId: number; - modelVersionId: number; - nsfw: boolean; - rating: number; - }) { - return this.track('resourceReviews', values); - } - - public reaction(values: { - type: ReactionType; - entityId: number; - ownerId: number; - reaction: ReviewReactions; - nsfw: NsfwLevelDeprecated; - }) { - return this.track('reactions', values); - } - - public question(values: { type: QuestionType; questionId: number }) { - return this.track('questions', values); - } - - public answer(values: { type: AnswerType; questionId: number; answerId: number }) { - return this.track('answers', values); - } - - public comment(values: { type: CommentType; entityId: number; nsfw: boolean }) { - return this.track('comments', values); - } - - public commentEvent(values: { type: CommentActivity; commentId: number }) { - return this.track('commentEvents', values); - } - - public post(values: { type: PostActivityType; postId: number; nsfw: boolean; tags: string[] }) { - return this.track('posts', values); - } - - public modelFile(values: { type: ModelFileActivity; id: number; modelVersionId: number }) { - return this.track('modelFileEvents', values); - } - - public images( - values: { - type: ImageActivityType; - imageId: number; - nsfw: NsfwLevelDeprecated; - tags: string[]; - ownerId: number; - tosReason?: string; - violationType?: string; - violationDetails?: string; - resources?: number[]; - userId?: number; - }[] - ) { - return this.trackMany('images', values); - } - - public bounty(values: { type: BountyActivity; bountyId: number; userId?: number }) { - return this.track('bounties', values); - } - - public bountyEntry(values: { - type: BountyEntryActivity; - bountyEntryId: number; - benefactorId?: number; - userId?: number; - }) { - return this.track('bountyEntries', values); - } - - public bountyBenefactor(values: { - type: BountyBenefactorActivity; - bountyId: number; - userId: number; - }) { - return this.track('bountyBenefactors', values); - } - - public modelEngagement(values: { type: ModelEngagementType; modelId: number }) { - return this.track('modelEngagements', values); - } - - public articleEngagement(values: { - type: ArticleEngagementType | `Delete${ArticleEngagementType}`; - articleId: number; - }) { - return this.track('articleEngagements', values); - } - - public tagEngagement(values: { type: TagEngagementType; tagId: number }) { - return this.track('tagEngagements', values); - } - - public userEngagement(values: { type: UserEngagementType; targetUserId: number }) { - return this.track('userEngagements', values); - } - - public bountyEngagement(values: { - type: BountyEngagementType | `Delete${BountyEngagementType}`; - bountyId: number; - }) { - return this.track('bountyEngagements', values); - } - - public prohibitedRequest(values: { - prompt: string; - negativePrompt: string; - source?: ProhibitedSources; - remixOfId?: number; - }) { - return this.track('prohibitedRequests', values); - } - - public report(values: { - type: ReportType; - entityType: string; - entityId: number; - reason: ReportReason; - status: ReportStatus; - }) { - return this.track('reports', values); - } - - public share(values: { url: string; platform: 'reddit' | 'twitter' | 'clipboard' }) { - return this.track('shares', values); - } - - public file(values: { type: FileActivity; entityType: string; entityId: number }) { - return this.track('files', values); - } - - public search(values: { query: string; index: string; filters?: any }) { - const { filters, ...rest } = values; - return this.track('search', { - ...rest, - filters: - filters != null ? (typeof filters === 'string' ? filters : JSON.stringify(filters)) : '', - }); - } - - public newOrderImageRating( - values: AddImageRatingInput & { - userId: number; - status: NewOrderImageRatingStatus; - grantedExp: number; - multiplier: number; - rank: NewOrderRankType; - originalLevel?: NsfwLevel; - voteWeight?: number; - } - ) { - return this.track('knights_new_order_image_rating', { ...values, createdAt: new Date() }); - } - - public entityMetric(values: { - entityType: EntityMetric_EntityType_Type; - entityId: number; - metricType: EntityMetric_MetricType_Type; - metricValue: number; - }) { - return this.track( - 'entityMetricEvents', - { ...values, createdAt: new Date() }, - { skipActorMeta: true } - ); - } - - public moderationRequest(values: { - entityType: AllModKeys; - entityId: number; - userId: number; - rules: string[]; - // value: string; - date: Date; - valid?: boolean; - }) { - return this.track('moderationRequest', { ...values }, { skipActorMeta: true }); - } - - public retoolAudit(values: { - action: string; - privileged: boolean; - outcome: 'ok' | 'error'; - errorMsg?: string; - payload: Record; - affected?: Record; - }) { - return this.track('retoolAuditLog', { - action: values.action, - privileged: values.privileged ? 1 : 0, - outcome: values.outcome, - errorMsg: values.errorMsg ?? '', - payload: JSON.stringify(values.payload), - affected: values.affected ? JSON.stringify(values.affected) : '', - }); - } + return client; } diff --git a/packages/civitai-clickhouse/src/env.ts b/packages/civitai-clickhouse/src/env.ts new file mode 100644 index 0000000000..c92c8e76c6 --- /dev/null +++ b/packages/civitai-clickhouse/src/env.ts @@ -0,0 +1,28 @@ +// Package-owned env schema for @civitai/clickhouse. Mirrors the clickhouse slice of +// the app's server-schema.ts (host/user/pass are required in prod, optional in dev). +import * as z from 'zod'; + +const isProd = process.env.NODE_ENV === 'production'; + +const schema = z.object({ + CLICKHOUSE_HOST: isProd ? z.string() : z.string().optional(), + CLICKHOUSE_USERNAME: isProd ? z.string() : z.string().optional(), + CLICKHOUSE_PASSWORD: isProd ? z.string() : z.string().optional(), +}); + +const parsed = schema.safeParse(process.env); +if (!parsed.success) { + throw new Error( + '[@civitai/clickhouse] Invalid environment variables:\n' + z.prettifyError(parsed.error) + ); +} + +// Normalized, env-derived defaults. The factory accepts a Partial. +export const clickhouseEnv = { + host: parsed.data.CLICKHOUSE_HOST, + username: parsed.data.CLICKHOUSE_USERNAME, + password: parsed.data.CLICKHOUSE_PASSWORD, + isProd, +}; + +export type ClickhouseConfig = typeof clickhouseEnv; diff --git a/packages/civitai-clickhouse/src/index.ts b/packages/civitai-clickhouse/src/index.ts new file mode 100644 index 0000000000..4f1cce44fa --- /dev/null +++ b/packages/civitai-clickhouse/src/index.ts @@ -0,0 +1 @@ +export * from './client'; diff --git a/packages/civitai-db-schema/package.json b/packages/civitai-db-schema/package.json new file mode 100644 index 0000000000..771a6d1128 --- /dev/null +++ b/packages/civitai-db-schema/package.json @@ -0,0 +1,7 @@ +{ + "name": "@civitai/db-schema", + "version": "0.0.0", + "private": true, + "main": "./src/index.ts", + "types": "./src/index.ts" +} diff --git a/packages/civitai-db-schema/prisma/schema.full.prisma b/packages/civitai-db-schema/prisma/schema.full.prisma index 0baeb92002..1bb9f2e57a 100644 --- a/packages/civitai-db-schema/prisma/schema.full.prisma +++ b/packages/civitai-db-schema/prisma/schema.full.prisma @@ -14,12 +14,12 @@ generator client { generator enums { provider = "node ./scripts/prisma-enum-generator.mjs" - output = "../src/shared/utils/prisma/enums.ts" + output = "../src/enums.ts" } generator typescriptInterfaces { provider = "prisma-generator-typescript-interfaces" - output = "../src/shared/utils/prisma/models.ts" + output = "../src/models.ts" } datasource db { diff --git a/packages/civitai-db-schema/src/index.ts b/packages/civitai-db-schema/src/index.ts new file mode 100644 index 0000000000..0d37223d4b --- /dev/null +++ b/packages/civitai-db-schema/src/index.ts @@ -0,0 +1,6 @@ +// Bare `@civitai/db-schema` exposes the generated Prisma client + types. +// The generated enum objects and model interfaces are intentionally NOT merged +// here (their names overlap); import them via subpaths instead: +// import { ... } from '@civitai/db-schema/enums'; +// import type { ... } from '@civitai/db-schema/models'; +export * from '@prisma/client'; diff --git a/packages/civitai-db/package.json b/packages/civitai-db/package.json new file mode 100644 index 0000000000..249cb03121 --- /dev/null +++ b/packages/civitai-db/package.json @@ -0,0 +1,10 @@ +{ + "name": "@civitai/db", + "version": "0.0.0", + "private": true, + "main": "./src/index.ts", + "types": "./src/index.ts", + "dependencies": { + "@civitai/db-schema": "workspace:*" + } +} diff --git a/packages/civitai-db/src/client.ts b/packages/civitai-db/src/client.ts index 62d0311813..7fcbaea756 100644 --- a/packages/civitai-db/src/client.ts +++ b/packages/civitai-db/src/client.ts @@ -1,100 +1,88 @@ -// src/server/db/client.ts -import type { Prisma } from '@prisma/client'; -import { PrismaClient } from '@prisma/client'; -import { isProd } from '~/env/other'; -import { env } from '~/env/server'; -import { logToAxiom } from '~/server/logging/client'; +// Prisma read/write client factory. The generated client + types come from the +// @civitai/db-schema contract package, never @prisma/client directly. +import type { Prisma } from '@civitai/db-schema'; +import { PrismaClient } from '@civitai/db-schema'; +import { dbEnv, type DbConfig } from './env'; -declare global { - // eslint-disable-next-line no-var, vars-on-top - var globalDbRead: PrismaClient | undefined; - // eslint-disable-next-line no-var, vars-on-top - var globalDbWrite: PrismaClient | undefined; -} +export type PrismaClients = { dbRead: PrismaClient; dbWrite: PrismaClient }; -const logFor = (target: 'write' | 'read') => - async function logQuery(e: { query: string; params: string; duration: number }) { - if (e.duration < 2000) return; - let query = e.query; - const params = JSON.parse(e.params); - // Replace $X variables with params in query so it's possible to copy/paste and optimize - for (let i = 0; i < params.length; i++) { - // Negative lookahead for no more numbers, ie. replace $1 in '$1' but not '$11' - const re = new RegExp('\\$' + ((i as number) + 1) + '(?!\\d)', 'g'); - // If string, will quote - if bool or numeric, will not - does the job here - if (typeof params[i] === 'string') params[i] = "'" + params[i].replace("'", "\\'") + "'"; - //params[i] = JSON.stringify(params[i]) - query = query.replace(re, params[i]); - } +export type CreatePrismaClientsOptions = Partial & { + /** Structured slow-query telemetry (the old logToAxiom call). Injected by the app. */ + onSlowQuery?: (e: { query: string; duration: number; target: 'read' | 'write' }) => void; +}; - if (!isProd) console.log(query); - else logToAxiom({ query, duration: e.duration, target }, 'db-logs'); - }; +/** + * Build the Prisma read/write clients. Connection config defaults come from the + * package env schema (./env, overridable via options); app behavior (logger, slow-query + * sink) is injected. HMR/global caching and the Next build guard live in the app shim + * that calls this. See the `~/server/db/client` shim. + */ +export function createPrismaClients(options: CreatePrismaClientsOptions = {}): PrismaClients { + const { onSlowQuery, ...envOverrides } = options; + const config = { ...dbEnv, ...envOverrides }; -const singleClient = env.DATABASE_REPLICA_URL === env.DATABASE_URL; -const createPrismaClient = ({ readonly }: { readonly: boolean }): PrismaClient => { - const log: Prisma.LogDefinition[] = env.LOGGING.filter((x) => x.startsWith('prisma:')).map( - (x) => ({ - emit: 'stdout', - level: x.replace('prisma:', '') as Prisma.LogLevel, - }) - ); - if (env.LOGGING.some((x) => x.includes('prisma-slow'))) { - const existingItemIndex = log.findIndex((x) => x.level === 'query'); - log.splice(existingItemIndex, 1); - log.push({ - emit: 'event', - level: 'query', - }); - } - const dbUrl = readonly ? env.DATABASE_REPLICA_URL : env.DATABASE_URL; - const options = { log, datasources: { db: { url: dbUrl } } } as Prisma.PrismaClientOptions; - const prisma = new PrismaClient(options); + const singleClient = config.replicaUrl === config.databaseUrl; - // use with prisma-slow,prisma-showparams - if (env.LOGGING.some((x) => x === 'prisma-showparams')) { - // @ts-ignore - prisma.$on('query', async (e: { query: string; params: string; duration: number }) => { + const logFor = (target: 'write' | 'read') => + async function logQuery(e: { query: string; params: string; duration: number }) { + if (e.duration < 2000) return; let query = e.query; const params = JSON.parse(e.params); - + // Replace $X variables with params in query so it's possible to copy/paste and optimize for (let i = 0; i < params.length; i++) { + // Negative lookahead for no more numbers, ie. replace $1 in '$1' but not '$11' const re = new RegExp('\\$' + ((i as number) + 1) + '(?!\\d)', 'g'); + // If string, will quote - if bool or numeric, will not - does the job here if (typeof params[i] === 'string') params[i] = "'" + params[i].replace("'", "\\'") + "'"; query = query.replace(re, params[i]); } - console.log(query); - }); - } - return prisma; -}; + if (!config.isProd) console.log(query); + else onSlowQuery?.({ query, duration: e.duration, target }); + }; -export let dbRead: PrismaClient; -export let dbWrite: PrismaClient; - -if (!env.IS_BUILD) { - if (isProd) { - dbWrite = createPrismaClient({ readonly: false }); - dbRead = singleClient ? dbWrite : createPrismaClient({ readonly: true }); - } else { - if (!global.globalDbWrite) { - global.globalDbWrite = createPrismaClient({ readonly: false }); - - if (env.LOGGING.includes('prisma-slow-write')) - // @ts-ignore - this is necessary to get the query event - global.globalDbWrite.$on('query', logFor('write')); + const createPrismaClient = ({ readonly }: { readonly: boolean }): PrismaClient => { + const logDef: Prisma.LogDefinition[] = config.logging + .filter((x) => x.startsWith('prisma:')) + .map((x) => ({ emit: 'stdout', level: x.replace('prisma:', '') as Prisma.LogLevel })); + if (config.logging.some((x) => x.includes('prisma-slow'))) { + const existingItemIndex = logDef.findIndex((x) => x.level === 'query'); + if (existingItemIndex >= 0) logDef.splice(existingItemIndex, 1); + logDef.push({ emit: 'event', level: 'query' }); } - if (!global.globalDbRead) { - global.globalDbRead = singleClient - ? global.globalDbWrite - : createPrismaClient({ readonly: true }); + const dbUrl = readonly ? config.replicaUrl : config.databaseUrl; + const clientOptions = { + log: logDef, + datasources: { db: { url: dbUrl } }, + } as Prisma.PrismaClientOptions; + const prisma = new PrismaClient(clientOptions); - if (env.LOGGING.includes('prisma-slow-read')) - // @ts-ignore - this is necessary to get the query event - global.globalDbRead.$on('query', logFor('read')); + // use with prisma-slow,prisma-showparams + if (config.logging.some((x) => x === 'prisma-showparams')) { + // @ts-ignore + prisma.$on('query', async (e: { query: string; params: string; duration: number }) => { + let query = e.query; + const params = JSON.parse(e.params); + for (let i = 0; i < params.length; i++) { + const re = new RegExp('\\$' + ((i as number) + 1) + '(?!\\d)', 'g'); + if (typeof params[i] === 'string') params[i] = "'" + params[i].replace("'", "\\'") + "'"; + query = query.replace(re, params[i]); + } + console.log(query); + }); } - dbWrite = global.globalDbWrite; - dbRead = singleClient ? dbWrite : global.globalDbRead; - } + return prisma; + }; + + const dbWrite = createPrismaClient({ readonly: false }); + const dbRead = singleClient ? dbWrite : createPrismaClient({ readonly: true }); + + if (config.logging.includes('prisma-slow-write')) + // @ts-ignore - necessary to get the query event + dbWrite.$on('query', logFor('write')); + if (config.logging.includes('prisma-slow-read')) + // @ts-ignore - necessary to get the query event + dbRead.$on('query', logFor('read')); + + return { dbRead, dbWrite }; } diff --git a/packages/civitai-db/src/concurrency-helpers.ts b/packages/civitai-db/src/concurrency-helpers.ts new file mode 100644 index 0000000000..65587e2e19 --- /dev/null +++ b/packages/civitai-db/src/concurrency-helpers.ts @@ -0,0 +1,68 @@ +// Vendored from src/server/utils/concurrency-helpers.ts (the `limitConcurrency` piece) +// so @civitai/db stays self-contained — no app-util import. + +export type Task = () => Promise; +type TaskGenerator = () => Task | null; + +function isTaskGenerator(arg: any): arg is TaskGenerator { + return typeof arg === 'function'; +} + +type LimitConcurrencyOptions = { + limit: number; + betweenTasksFn?: () => Promise; +}; + +export function limitConcurrency( + tasksOrGenerator: Task[] | TaskGenerator, + options?: LimitConcurrencyOptions | number +): Promise { + if (typeof options === 'number') options = { limit: options } as LimitConcurrencyOptions; + if (!options) options = { limit: 1 } as LimitConcurrencyOptions; + const { limit, betweenTasksFn } = options; + + return new Promise((resolve, reject) => { + let active = 0; + let finished = false; + let index = 0; + const isGenerator = isTaskGenerator(tasksOrGenerator); + const tasks = isGenerator ? [] : (tasksOrGenerator as Task[]); + + const getNextTask = async (): Promise => { + if (betweenTasksFn) await betweenTasksFn(); + if (isGenerator) return tasksOrGenerator(); + else { + if (index < tasks.length) return tasks[index++]; + return null; + } + }; + + const checkFinished = () => { + if (finished && active === 0) resolve(); + }; + + const run = async () => { + const task = await getNextTask(); + if (!task) { + finished = true; + checkFinished(); + return; + } + + active++; + try { + await task(); + } catch (error) { + reject(error); + return; + } finally { + active--; + checkFinished(); + if (active < limit && !finished) run(); // Start a new task if we're below the concurrency limit + } + }; + + // Start the initial set of tasks + for (let i = 0; i < limit; i++) run(); + }); +} diff --git a/packages/civitai-db/src/db-helpers.ts b/packages/civitai-db/src/db-helpers.ts index a6b548f839..ed59a2a1cd 100644 --- a/packages/civitai-db/src/db-helpers.ts +++ b/packages/civitai-db/src/db-helpers.ts @@ -1,12 +1,15 @@ -import { Prisma } from '@prisma/client'; +import { Prisma } from '@civitai/db-schema'; import type { QueryResult, QueryResultRow } from 'pg'; -import { Pool } from 'pg'; +import { Pool, types } from 'pg'; import { performance } from 'node:perf_hooks'; import client from 'prom-client'; -import { env } from '~/env/server'; -import { dbWrite } from '~/server/db/client'; -import { limitConcurrency } from '~/server/utils/concurrency-helpers'; -import { createLogger } from '~/utils/logging'; +import { dbEnv, type DbConfig, type DbLogFn } from './env'; +import { limitConcurrency } from './concurrency-helpers'; + +// Fix Dates: TIMESTAMP comes back as a UTC Date (was set per-pool-module in the app). +types.setTypeParser(types.builtins.TIMESTAMP, function (stringValue) { + return new Date(stringValue.replace(' ', 'T') + 'Z'); +}); // Histogram for pg Pool acquire latency. Defined here (not in prom/client.ts) // to avoid a module-init cycle: prom/client.ts imports pgDb/notifDb/datapacketDb, @@ -35,8 +38,6 @@ const pgPoolAcquireHistogram = (() => { } })(); -const log = createLogger('pgDb', 'blue'); - /** * Formats a value for SQL display/logging. * Used by combineSqlWithParams for consistent value formatting. @@ -73,25 +74,31 @@ type ClientInstanceType = | 'notification' | 'notificationRead' | 'datapacketRead'; -const instanceUrlMap: Record = { - notification: env.NOTIFICATION_DB_URL, - notificationRead: env.NOTIFICATION_DB_REPLICA_URL ?? env.NOTIFICATION_DB_URL, - primary: env.DATABASE_URL, - primaryRead: env.DATABASE_REPLICA_URL ?? env.DATABASE_URL, - primaryReadLong: env.DATABASE_REPLICA_LONG_URL ?? env.DATABASE_URL, - datapacketRead: env.DATAPACKET_DATABASE_RO_URL ?? env.DATABASE_URL, +export type GetClientOptions = Partial & { + instance?: ClientInstanceType; + /** Debug logger (app-defined). Defaults to a no-op. */ + log?: DbLogFn; }; -export function getClient( - { instance }: { instance: ClientInstanceType } = { - instance: 'primary', - } -) { +export function getClient(options: GetClientOptions = {}) { + const { instance = 'primary', log: logOption, ...envOverrides } = options; + const config = { ...dbEnv, ...envOverrides }; + const log: DbLogFn = logOption ?? (() => {}); + + const instanceUrlMap: Record = { + notification: config.notificationUrl, + notificationRead: config.notificationReplicaUrl ?? config.notificationUrl, + primary: config.databaseUrl, + primaryRead: config.replicaUrl ?? config.databaseUrl, + primaryReadLong: config.replicaLongUrl ?? config.databaseUrl, + datapacketRead: config.datapacketReadUrl ?? config.databaseUrl, + }; + log(`Creating ${instance} client`); const envUrl = instanceUrlMap[instance]; const connectionStringUrl = new URL(envUrl); - if (env.DATABASE_SSL !== false) connectionStringUrl.searchParams.set('sslmode', 'no-verify'); + if (config.ssl !== false) connectionStringUrl.searchParams.set('sslmode', 'no-verify'); const connectionString = connectionStringUrl.toString(); const isNotification = instance === 'notification' || instance === 'notificationRead'; @@ -105,40 +112,40 @@ export function getClient( // For notification instances, we set it per-connection via SET instead. const notifStatementTimeout = instance === 'notificationRead' - ? (env.IS_DATAPACKET ? env.DATABASE_READ_TIMEOUT ?? 10000 : undefined) + ? (config.isDatapacket ? config.readTimeout ?? 10000 : undefined) : instance === 'notification' - ? env.DATABASE_WRITE_TIMEOUT + ? config.writeTimeout : undefined; const pool = new Pool({ connectionString, - connectionTimeoutMillis: env.IS_DATAPACKET - ? env.DATABASE_CONNECTION_TIMEOUT || 5000 - : env.DATABASE_CONNECTION_TIMEOUT, + connectionTimeoutMillis: config.isDatapacket + ? config.connectionTimeout || 5000 + : config.connectionTimeout, min: 0, - max: isNotification ? (env.NOTIFICATION_POOL_MAX ?? env.DATABASE_POOL_MAX) : env.DATABASE_POOL_MAX, + max: isNotification ? (config.notificationPoolMax ?? config.poolMax) : config.poolMax, // trying this for leaderboard job - idleTimeoutMillis: instance === 'primaryReadLong' ? 300_000 : env.DATABASE_POOL_IDLE_TIMEOUT, + idleTimeoutMillis: instance === 'primaryReadLong' ? 300_000 : config.poolIdleTimeout, statement_timeout: - (isNotification || instance === 'datapacketRead') && env.IS_DATAPACKET + (isNotification || instance === 'datapacketRead') && config.isDatapacket ? undefined // DP: set per-connection below (PgBouncer ignores startup params) : instance === 'notificationRead' ? undefined // DOKS: standby doesn't support this : instance === 'primaryRead' - ? env.DATABASE_READ_TIMEOUT - : env.DATABASE_WRITE_TIMEOUT, - application_name: `${appBaseName}${env.PODNAME ? '-' + env.PODNAME : ''}`, + ? config.readTimeout + : config.writeTimeout, + application_name: `${appBaseName}${config.podName ? '-' + config.podName : ''}`, }) as AugmentedPool; // Set statement_timeout per-connection on DP instances that go through PgBouncer // (PgBouncer ignores statement_timeout as a startup parameter) - if (env.IS_DATAPACKET && isNotification && notifStatementTimeout) { + if (config.isDatapacket && isNotification && notifStatementTimeout) { pool.on('connect', (client) => { client.query(`SET statement_timeout = ${Number(notifStatementTimeout)}`).catch(() => {}); }); } - if (env.IS_DATAPACKET && instance === 'datapacketRead') { - const readTimeout = env.DATABASE_READ_TIMEOUT ?? 120000; // 2 minutes default + if (config.isDatapacket && instance === 'datapacketRead') { + const readTimeout = config.readTimeout ?? 120000; // 2 minutes default pool.on('connect', (client) => { client.query(`SET statement_timeout = ${Number(readTimeout)}`).catch(() => {}); }); @@ -334,36 +341,8 @@ export function parameterizedTemplateHandler( }; } -function lsnGTE(lsn1: string, lsn2: string): boolean { - const [a1, b1] = lsn1.split('/').map((part) => parseInt(part, 16)); - const [a2, b2] = lsn2.split('/').map((part) => parseInt(part, 16)); - return a1 > a2 || (a1 === a2 && b1 >= b2); -} - -export async function getCurrentLSN() { - try { - const currentRes = await dbWrite.$queryRaw< - { - lsn: string; - }[] - >`SELECT pg_current_wal_lsn()::text AS lsn`; - return currentRes[0]?.lsn ?? ''; - } catch (e) { - // TODO what to return here - return ''; - } -} - -export async function checkNotUpToDate(lsn: string) { - try { - const roRes = await dbWrite.$queryRaw< - { replay_lsn: string }[] - >`SELECT replay_lsn::text FROM get_replication_status() where application_name like 'ro-c16-%'`; - return roRes.some((row) => !lsnGTE(row.replay_lsn, lsn)); - } catch (e) { - return true; - } -} +// getCurrentLSN / checkNotUpToDate / dbKV moved to ./kv-helpers (they need a Prisma +// client; the app shim binds dbWrite and re-exports them under the same names). export type RunContext = { cancelFns: (() => Promise)[]; @@ -535,19 +514,3 @@ export function getExplainSql(value: typeof Prisma.Sql) { export function jsonbArrayFrom(data: any): string { return `'${JSON.stringify(data)}'::jsonb`; } - -export const dbKV = { - get: async function (key: string, defaultValue?: T) { - const stored = await dbWrite.keyValue.findUnique({ where: { key } }); - return stored ? (stored.value as T) : defaultValue; - }, - set: async function (key: string, value: T) { - const json = JSON.stringify(value).replace(/'/g, "''"); - await dbWrite.$executeRawUnsafe(` - INSERT INTO "KeyValue" ("key", "value") - VALUES ('${key}', '${json}'::jsonb) - ON CONFLICT ("key") - DO UPDATE SET "value" = '${json}'::jsonb - `); - }, -}; diff --git a/packages/civitai-db/src/env.ts b/packages/civitai-db/src/env.ts new file mode 100644 index 0000000000..185cc55853 --- /dev/null +++ b/packages/civitai-db/src/env.ts @@ -0,0 +1,63 @@ +// Package-owned env schema for @civitai/db. Mirrors the postgres slice of the app's +// server-schema.ts so any app validates the same vars the same way on deployment. +import * as z from 'zod'; + +const booleanString = z.preprocess((val) => val === true || val === 'true', z.boolean()); +const commaDelimitedStringArray = z.preprocess((val) => { + if (Array.isArray(val)) return val; + if (typeof val === 'string') + return val + .split(',') + .map((s) => s.trim()) + .filter(Boolean); + return []; +}, z.string().array()); + +const schema = z.object({ + DATABASE_URL: z.url(), + DATABASE_REPLICA_URL: z.url(), + DATABASE_REPLICA_LONG_URL: z.url().optional(), + DATABASE_SSL: booleanString.default(true), + NOTIFICATION_DB_URL: z.url(), + NOTIFICATION_DB_REPLICA_URL: z.url(), + DATAPACKET_DATABASE_RO_URL: z.url().optional(), + DATABASE_CONNECTION_TIMEOUT: z.coerce.number().default(0), + DATABASE_POOL_MAX: z.coerce.number().default(20), + NOTIFICATION_POOL_MAX: z.coerce.number().optional(), + DATABASE_POOL_IDLE_TIMEOUT: z.coerce.number().default(30000), + DATABASE_READ_TIMEOUT: z.coerce.number().optional(), + DATABASE_WRITE_TIMEOUT: z.coerce.number().optional(), + IS_DATAPACKET: booleanString.default(false), + PODNAME: z.string().optional(), + LOGGING: commaDelimitedStringArray, +}); + +const parsed = schema.safeParse(process.env); +if (!parsed.success) { + throw new Error('[@civitai/db] Invalid environment variables:\n' + z.prettifyError(parsed.error)); +} + +// Normalized, env-derived defaults. Factories accept a Partial to override. +export const dbEnv = { + databaseUrl: parsed.data.DATABASE_URL, + replicaUrl: parsed.data.DATABASE_REPLICA_URL, + replicaLongUrl: parsed.data.DATABASE_REPLICA_LONG_URL, + ssl: parsed.data.DATABASE_SSL, + notificationUrl: parsed.data.NOTIFICATION_DB_URL, + notificationReplicaUrl: parsed.data.NOTIFICATION_DB_REPLICA_URL, + datapacketReadUrl: parsed.data.DATAPACKET_DATABASE_RO_URL, + connectionTimeout: parsed.data.DATABASE_CONNECTION_TIMEOUT, + poolMax: parsed.data.DATABASE_POOL_MAX, + notificationPoolMax: parsed.data.NOTIFICATION_POOL_MAX, + poolIdleTimeout: parsed.data.DATABASE_POOL_IDLE_TIMEOUT, + readTimeout: parsed.data.DATABASE_READ_TIMEOUT, + writeTimeout: parsed.data.DATABASE_WRITE_TIMEOUT, + isDatapacket: parsed.data.IS_DATAPACKET, + podName: parsed.data.PODNAME, + logging: parsed.data.LOGGING, + // NODE_ENV is a universal Node convention; the Next build guard lives in the app shim. + isProd: process.env.NODE_ENV === 'production', +}; + +export type DbConfig = typeof dbEnv; +export type DbLogFn = (message: string, ...args: unknown[]) => void; diff --git a/packages/civitai-db/src/index.ts b/packages/civitai-db/src/index.ts new file mode 100644 index 0000000000..6d32d90444 --- /dev/null +++ b/packages/civitai-db/src/index.ts @@ -0,0 +1,4 @@ +export * from './env'; +export * from './client'; +export * from './db-helpers'; +export * from './kv-helpers'; diff --git a/packages/civitai-db/src/kv-helpers.ts b/packages/civitai-db/src/kv-helpers.ts new file mode 100644 index 0000000000..fad172bdf5 --- /dev/null +++ b/packages/civitai-db/src/kv-helpers.ts @@ -0,0 +1,49 @@ +// Helpers that need a Prisma client. Kept separate from db-helpers.ts so the package +// has no client.ts <-> db-helpers.ts cycle: callers pass `dbWrite` in. The app shim +// binds its dbWrite and re-exports these under their original names/signatures. +import type { PrismaClient } from '@civitai/db-schema'; + +function lsnGTE(lsn1: string, lsn2: string): boolean { + const [a1, b1] = lsn1.split('/').map((part) => parseInt(part, 16)); + const [a2, b2] = lsn2.split('/').map((part) => parseInt(part, 16)); + return a1 > a2 || (a1 === a2 && b1 >= b2); +} + +export async function getCurrentLSN(dbWrite: PrismaClient) { + try { + const currentRes = await dbWrite.$queryRaw<{ lsn: string }[]>`SELECT pg_current_wal_lsn()::text AS lsn`; + return currentRes[0]?.lsn ?? ''; + } catch (e) { + // TODO what to return here + return ''; + } +} + +export async function checkNotUpToDate(dbWrite: PrismaClient, lsn: string) { + try { + const roRes = await dbWrite.$queryRaw< + { replay_lsn: string }[] + >`SELECT replay_lsn::text FROM get_replication_status() where application_name like 'ro-c16-%'`; + return roRes.some((row) => !lsnGTE(row.replay_lsn, lsn)); + } catch (e) { + return true; + } +} + +export function makeDbKV(dbWrite: PrismaClient) { + return { + get: async function (key: string, defaultValue?: T) { + const stored = await dbWrite.keyValue.findUnique({ where: { key } }); + return stored ? (stored.value as T) : defaultValue; + }, + set: async function (key: string, value: T) { + const json = JSON.stringify(value).replace(/'/g, "''"); + await dbWrite.$executeRawUnsafe(` + INSERT INTO "KeyValue" ("key", "value") + VALUES ('${key}', '${json}'::jsonb) + ON CONFLICT ("key") + DO UPDATE SET "value" = '${json}'::jsonb + `); + }, + }; +} diff --git a/packages/civitai-redis/package.json b/packages/civitai-redis/package.json new file mode 100644 index 0000000000..872dbf1895 --- /dev/null +++ b/packages/civitai-redis/package.json @@ -0,0 +1,7 @@ +{ + "name": "@civitai/redis", + "version": "0.0.0", + "private": true, + "main": "./src/index.ts", + "types": "./src/index.ts" +} diff --git a/packages/civitai-redis/src/client.ts b/packages/civitai-redis/src/client.ts index ec6a2df223..5fb66072d3 100644 --- a/packages/civitai-redis/src/client.ts +++ b/packages/civitai-redis/src/client.ts @@ -3,11 +3,16 @@ import { pack, unpack } from 'msgpackr'; import type { RedisClientType, SetOptions } from 'redis'; import { createClient, createCluster } from 'redis'; import { RESP_TYPES } from 'redis'; -import { isProd } from '~/env/other'; -import { env } from '~/env/server'; -import { createLogger } from '~/utils/logging'; -import { slugit } from '~/utils/string-helpers'; -import { FLIPT_FEATURE_FLAGS, isFlipt } from '~/server/flipt/client'; +import slugify from 'slugify'; +import { redisEnv, type RedisConfig } from './env'; + +export type { RedisConfig } from './env'; +export type RedisLogFn = (message: string, ...args: unknown[]) => void; +/** Resolves whether enhanced cluster failover is enabled — injected app policy (Flipt). */ +export type RedisFailoverResolver = (context: Record) => Promise; + +// inlined — was slugit from ~/utils/string-helpers +const slugit = (value: string) => slugify(value, { lower: true, strict: true }); export type RedisKeyStringsCache = Values; export type RedisKeyStringsSys = Values; @@ -176,14 +181,12 @@ interface CustomRedisClientCache extends CustomRedisClient {} -declare global { - // eslint-disable-next-line no-var, vars-on-top - var globalRedis: CustomRedisClientCache | undefined; - // eslint-disable-next-line no-var, vars-on-top - var globalSysRedis: CustomRedisClientSys | undefined; -} - -const log = createLogger('redis', 'green'); +// Configured once per process by createRedisClients(). The defaults keep the +// module-level helpers safe if referenced before the factory runs. (HMR/global +// singleton caching lives in the app shim, where the factory is called.) +let config: RedisConfig = redisEnv; +let log: RedisLogFn = () => {}; +let isEnhancedFailoverEnabled: RedisFailoverResolver = async () => false; // Track topology refresh intervals for cleanup const clusterRefreshIntervals = new Map>(); @@ -230,8 +233,8 @@ function triggerTopologyRediscovery(clusterClient: any, reason: string) { * Falls back to single URL if REDIS_CLUSTER_NODES is not set. */ function parseClusterNodes(fallbackUrl: string): { url: string }[] { - if (env.REDIS_CLUSTER_NODES) { - const nodes = env.REDIS_CLUSTER_NODES.split(',') + if (config.clusterNodes) { + const nodes = config.clusterNodes.split(',') .map((nodeUrl) => nodeUrl.trim()) .filter(Boolean) .map((nodeUrl) => { @@ -252,7 +255,7 @@ function parseClusterNodes(fallbackUrl: string): { url: string }[] { function getBaseClient(type: 'cache' | 'system') { log(`Creating Redis client (${type})`); - const REDIS_URL = type === 'system' ? env.REDIS_SYS_URL : env.REDIS_URL; + const REDIS_URL = type === 'system' ? config.sysUrl : config.url; const url = new URL(REDIS_URL); const connectionUrl = `${url.protocol}//${url.host}`; @@ -262,7 +265,7 @@ function getBaseClient(type: 'cache' | 'system') { log(`Redis reconnecting, retry ${retries}`); return Math.min(retries * 100, 3000); }, - connectTimeout: env.REDIS_TIMEOUT, + connectTimeout: config.timeout, }; const authConfig = { @@ -274,7 +277,7 @@ function getBaseClient(type: 'cache' | 'system') { // System redis is always a single node // Cache redis can be either cluster or single node based on env.REDIS_CLUSTER - const isCluster = type === 'cache' && env.REDIS_CLUSTER; + const isCluster = type === 'cache' && config.cluster; const baseClient = isCluster ? createCluster({ @@ -310,7 +313,7 @@ function getBaseClient(type: 'cache' | 'system') { const getFliptHostname = (): string => { try { // NEXTAUTH_URL contains the full URL like https://next.civitai.com - const nextAuthUrl = env.NEXTAUTH_URL; + const nextAuthUrl = config.nextAuthUrl; if (nextAuthUrl) { return new URL(nextAuthUrl).hostname; } @@ -321,21 +324,16 @@ function getBaseClient(type: 'cache' | 'system') { }; const fliptHostname = getFliptHostname(); const fliptContext: Record = { hostname: fliptHostname }; - if (env.FLIPT_DEPLOYMENT_ID) fliptContext.deploymentId = env.FLIPT_DEPLOYMENT_ID; + if (config.fliptDeploymentId) fliptContext.deploymentId = config.fliptDeploymentId; log( `Flipt context for enhanced failover flag: hostname=${fliptHostname}, deploymentId=${ - env.FLIPT_DEPLOYMENT_ID ?? 'unset' + config.fliptDeploymentId ?? 'unset' }` ); - // Helper to check feature flag before triggering rediscovery + // Helper to check the injected app policy before triggering rediscovery const maybeRediscover = async (reason: string) => { - const isEnhancedFailoverEnabled = await isFlipt( - FLIPT_FEATURE_FLAGS.REDIS_CLUSTER_ENHANCED_FAILOVER, - 'redis-cluster', // entityId - fliptContext // context for segment matching - ); - if (isEnhancedFailoverEnabled) { + if (await isEnhancedFailoverEnabled(fliptContext)) { triggerTopologyRediscovery(baseClient, reason); } }; @@ -376,20 +374,16 @@ function getBaseClient(type: 'cache' | 'system') { const setupEnhancedFailover = async () => { try { log('Checking enhanced failover feature flag...'); - const isEnhancedFailoverEnabled = await isFlipt( - FLIPT_FEATURE_FLAGS.REDIS_CLUSTER_ENHANCED_FAILOVER, - 'redis-cluster', // entityId - fliptContext // context for segment matching - ); + const enhancedFailoverEnabled = await isEnhancedFailoverEnabled(fliptContext); - if (!isEnhancedFailoverEnabled) { + if (!enhancedFailoverEnabled) { log('Enhanced cluster failover handling is DISABLED (feature flag off)'); return; } log('Enhanced cluster failover handling is ENABLED'); - const refreshInterval = env.REDIS_CLUSTER_REFRESH_INTERVAL; + const refreshInterval = config.clusterRefreshInterval; if (refreshInterval > 0) { const intervalId = setInterval(() => { triggerTopologyRediscovery(baseClient, 'periodic refresh'); @@ -656,21 +650,31 @@ function getSysClient() { return getClient('system') as CustomRedisClientSys; } -export let redis: CustomRedisClientCache; -export let sysRedis: CustomRedisClientSys; -if (!env.IS_BUILD) { - if (isProd) { - redis = getCacheClient(); - sysRedis = getSysClient(); - } else { - if (!global.globalRedis) global.globalRedis = getCacheClient(); - redis = global.globalRedis; +export type RedisClients = { redis: CustomRedisClientCache; sysRedis: CustomRedisClientSys }; - if (!global.globalSysRedis) global.globalSysRedis = getSysClient(); - sysRedis = global.globalSysRedis; - } -} else { - log('Skipping Redis initialization (build phase)'); +export type CreateRedisClientsOptions = Partial & { + /** Debug logger (app-defined). Defaults to a no-op. */ + log?: RedisLogFn; + /** + * App policy resolving whether enhanced cluster failover is enabled for a given + * Flipt context. Defaults to always-off; the app shim wires the real Flipt call. + */ + isEnhancedFailoverEnabled?: RedisFailoverResolver; +}; + +/** + * Build the cache + system Redis clients. Connection config defaults come from the + * package env schema (./env, overridable via options); app behavior (logger, failover + * policy) is injected. HMR/global singleton caching and the Next build guard live in + * the app shim that calls this. See the `~/server/redis/client` shim. + */ +export function createRedisClients(options: CreateRedisClientsOptions = {}): RedisClients { + const { log: logOption, isEnhancedFailoverEnabled: failoverOption, ...envOverrides } = options; + config = { ...redisEnv, ...envOverrides }; + if (logOption) log = logOption; + if (failoverOption) isEnhancedFailoverEnabled = failoverOption; + + return { redis: getCacheClient(), sysRedis: getSysClient() }; } // Source of Truth data diff --git a/packages/civitai-redis/src/env.ts b/packages/civitai-redis/src/env.ts new file mode 100644 index 0000000000..9f244a64f2 --- /dev/null +++ b/packages/civitai-redis/src/env.ts @@ -0,0 +1,37 @@ +// Package-owned env schema for @civitai/redis. Mirrors the redis slice of the app's +// server-schema.ts so any app validates the same vars the same way on deployment. +import * as z from 'zod'; + +const schema = z.object({ + REDIS_URL: z.url(), + REDIS_SYS_URL: z.url(), + REDIS_TIMEOUT: z.preprocess((x) => (x ? parseInt(String(x)) : 5000), z.number().optional()), + REDIS_CLUSTER: z.preprocess((x) => x === 'true', z.boolean().default(false)), + // Comma-separated list of cluster node URLs for redundant discovery + REDIS_CLUSTER_NODES: z.string().optional(), + // Topology refresh interval in ms (default 30s) + REDIS_CLUSTER_REFRESH_INTERVAL: z.coerce.number().default(30000), + // Used only to derive a hostname for the failover feature-flag context + NEXTAUTH_URL: z.string().optional(), + FLIPT_DEPLOYMENT_ID: z.string().optional(), +}); + +const parsed = schema.safeParse(process.env); +if (!parsed.success) { + throw new Error('[@civitai/redis] Invalid environment variables:\n' + z.prettifyError(parsed.error)); +} + +// Normalized, env-derived defaults. The factory accepts a Partial to +// override any of these per call. +export const redisEnv = { + url: parsed.data.REDIS_URL, + sysUrl: parsed.data.REDIS_SYS_URL, + timeout: parsed.data.REDIS_TIMEOUT, + cluster: parsed.data.REDIS_CLUSTER, + clusterNodes: parsed.data.REDIS_CLUSTER_NODES, + clusterRefreshInterval: parsed.data.REDIS_CLUSTER_REFRESH_INTERVAL, + nextAuthUrl: parsed.data.NEXTAUTH_URL, + fliptDeploymentId: parsed.data.FLIPT_DEPLOYMENT_ID, +}; + +export type RedisConfig = typeof redisEnv; diff --git a/packages/civitai-redis/src/index.ts b/packages/civitai-redis/src/index.ts new file mode 100644 index 0000000000..4f1cce44fa --- /dev/null +++ b/packages/civitai-redis/src/index.ts @@ -0,0 +1 @@ +export * from './client'; diff --git a/packages/civitai-telemetry/package.json b/packages/civitai-telemetry/package.json new file mode 100644 index 0000000000..36bbfaa4f6 --- /dev/null +++ b/packages/civitai-telemetry/package.json @@ -0,0 +1,7 @@ +{ + "name": "@civitai/telemetry", + "version": "0.0.0", + "private": true, + "main": "./src/index.ts", + "types": "./src/index.ts" +} diff --git a/packages/civitai-telemetry/src/client.ts b/packages/civitai-telemetry/src/client.ts index f1dea31cc2..909e366f09 100644 --- a/packages/civitai-telemetry/src/client.ts +++ b/packages/civitai-telemetry/src/client.ts @@ -1,8 +1,5 @@ import type { Counter, Gauge, Histogram } from 'prom-client'; import client from 'prom-client'; -import { datapacketDbRead } from '~/server/db/datapacketDb'; -import { notifDbRead, notifDbWrite } from '~/server/db/notifDb'; -import { pgDbRead, pgDbReadLong, pgDbWrite } from '~/server/db/pgDb'; const PROM_PREFIX = 'civitai_app_'; export function registerCounter({ name, help }: { name: string; help: string }) { @@ -197,100 +194,5 @@ export const dbReadFallbackCounter = registerCounterWithLabels({ labelNames: ['entity', 'caller'] as const, }); -// pgPoolAcquireHistogram is registered in src/server/db/db-helpers.ts, not here. -// Defining it here would create a module-init cycle (prom/client.ts imports -// pgDb → db-helpers, which would import this histogram back), which webpack's -// CJS chunking can break with a TDZ error at runtime. - -declare global { - // eslint-disable-next-line no-var - var pgGaugeInitialized: boolean; -} - -if (!global.pgGaugeInitialized) { - new client.Gauge({ - name: 'node_postgres_read_total_count', - help: 'node postgres read total count', - collect() { - this.set(pgDbRead.totalCount); - }, - }); - new client.Gauge({ - name: 'node_postgres_read_idle_count', - help: 'node postgres read idle count', - collect() { - this.set(pgDbRead.idleCount); - }, - }); - new client.Gauge({ - name: 'node_postgres_read_waiting_count', - help: 'node postgres read waiting count', - collect() { - this.set(pgDbRead.waitingCount); - }, - }); - new client.Gauge({ - name: 'node_postgres_write_total_count', - help: 'node postgres write total count', - collect() { - this.set(pgDbWrite.totalCount); - }, - }); - new client.Gauge({ - name: 'node_postgres_write_idle_count', - help: 'node postgres write idle count', - collect() { - this.set(pgDbWrite.idleCount); - }, - }); - new client.Gauge({ - name: 'node_postgres_write_waiting_count', - help: 'node postgres write waiting count', - collect() { - this.set(pgDbWrite.waitingCount); - }, - }); - - // Labeled pool metrics for all pools - new client.Gauge({ - name: 'node_postgres_pool_total_count', - help: 'Total connections in pg pool', - labelNames: ['pool'], - collect() { - this.set({ pool: 'read' }, pgDbRead?.totalCount ?? 0); - this.set({ pool: 'write' }, pgDbWrite?.totalCount ?? 0); - this.set({ pool: 'read_long' }, pgDbReadLong?.totalCount ?? 0); - this.set({ pool: 'notif_read' }, notifDbRead?.totalCount ?? 0); - this.set({ pool: 'notif_write' }, notifDbWrite?.totalCount ?? 0); - this.set({ pool: 'datapacket_read' }, datapacketDbRead?.totalCount ?? 0); - }, - }); - new client.Gauge({ - name: 'node_postgres_pool_idle_count', - help: 'Idle connections in pg pool', - labelNames: ['pool'], - collect() { - this.set({ pool: 'read' }, pgDbRead?.idleCount ?? 0); - this.set({ pool: 'write' }, pgDbWrite?.idleCount ?? 0); - this.set({ pool: 'read_long' }, pgDbReadLong?.idleCount ?? 0); - this.set({ pool: 'notif_read' }, notifDbRead?.idleCount ?? 0); - this.set({ pool: 'notif_write' }, notifDbWrite?.idleCount ?? 0); - this.set({ pool: 'datapacket_read' }, datapacketDbRead?.idleCount ?? 0); - }, - }); - new client.Gauge({ - name: 'node_postgres_pool_waiting_count', - help: 'Waiting connections in pg pool', - labelNames: ['pool'], - collect() { - this.set({ pool: 'read' }, pgDbRead?.waitingCount ?? 0); - this.set({ pool: 'write' }, pgDbWrite?.waitingCount ?? 0); - this.set({ pool: 'read_long' }, pgDbReadLong?.waitingCount ?? 0); - this.set({ pool: 'notif_read' }, notifDbRead?.waitingCount ?? 0); - this.set({ pool: 'notif_write' }, notifDbWrite?.waitingCount ?? 0); - this.set({ pool: 'datapacket_read' }, datapacketDbRead?.waitingCount ?? 0); - }, - }); - - global.pgGaugeInitialized = true; -} +// NOTE: the DB pool-depth gauges live in the app (src/server/prom/client.ts) — they +// compose the db pools + these prom helpers, which is app-level glue, not infra. diff --git a/packages/civitai-telemetry/src/index.ts b/packages/civitai-telemetry/src/index.ts new file mode 100644 index 0000000000..81815e2036 --- /dev/null +++ b/packages/civitai-telemetry/src/index.ts @@ -0,0 +1,2 @@ +export * from './client'; +export * from './otel-helpers'; diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml index 4c12aeca4e..04487e9a75 100644 --- a/pnpm-lock.yaml +++ b/pnpm-lock.yaml @@ -798,6 +798,22 @@ importers: specifier: ^8.19.0 version: 8.19.0(bufferutil@4.0.9)(utf-8-validate@5.0.10) + packages/civitai-axiom: {} + + packages/civitai-clickhouse: {} + + packages/civitai-db: + dependencies: + '@civitai/db-schema': + specifier: workspace:* + version: link:../civitai-db-schema + + packages/civitai-db-schema: {} + + packages/civitai-redis: {} + + packages/civitai-telemetry: {} + packages: '@acemir/cssom@0.9.31': diff --git a/pnpm-workspace.yaml b/pnpm-workspace.yaml new file mode 100644 index 0000000000..4f14504dc0 --- /dev/null +++ b/pnpm-workspace.yaml @@ -0,0 +1,4 @@ +packages: + - '.' + - 'packages/*' + - 'apps/*' diff --git a/scripts/generate-slim-schema.js b/scripts/generate-slim-schema.js index a6d2a7c727..8afe9c8618 100644 --- a/scripts/generate-slim-schema.js +++ b/scripts/generate-slim-schema.js @@ -11,8 +11,14 @@ const fs = require('fs'); const path = require('path'); -const FULL_SCHEMA_PATH = path.join(__dirname, '../prisma/schema.full.prisma'); -const SLIM_SCHEMA_PATH = path.join(__dirname, '../prisma/schema.prisma'); +const FULL_SCHEMA_PATH = path.join( + __dirname, + '../packages/civitai-db-schema/prisma/schema.full.prisma' +); +const SLIM_SCHEMA_PATH = path.join( + __dirname, + '../packages/civitai-db-schema/prisma/schema.prisma' +); function generateSlimSchema() { console.log('📋 Reading full schema from:', FULL_SCHEMA_PATH); diff --git a/src/server/clickhouse/client.ts b/src/server/clickhouse/client.ts new file mode 100644 index 0000000000..22959cd77e --- /dev/null +++ b/src/server/clickhouse/client.ts @@ -0,0 +1,27 @@ +// App shim for @civitai/clickhouse. The package owns the base client + env schema; the +// app injects the debug logger, owns the HMR singleton + Next build guard, and re-exports +// the base client surface plus the app-side Tracker (./tracker) for existing call sites. +import { createClickhouseClient, type CustomClickHouseClient } from '@civitai/clickhouse/client'; +import { isProd } from '~/env/other'; +import { env } from '~/env/server'; +import { createLogger } from '~/utils/logging'; + +export * from '@civitai/clickhouse/client'; + +declare global { + // eslint-disable-next-line no-var, vars-on-top + var globalClickhouse: CustomClickHouseClient | undefined; +} + +const make = () => createClickhouseClient({ log: createLogger('clickhouse', 'blue') }); + +const shouldConnect = !env.IS_BUILD && env.CLICKHOUSE_HOST && env.CLICKHOUSE_USERNAME; +export const clickhouse: CustomClickHouseClient | undefined = !shouldConnect + ? undefined + : isProd + ? make() + : (global.globalClickhouse ??= make()); + +// The Tracker is app-coupled (auth/session/schemas); it lives in the app and is +// re-exported here so existing `~/server/clickhouse/client` imports keep working. +export * from './tracker'; diff --git a/src/server/clickhouse/tracker.ts b/src/server/clickhouse/tracker.ts new file mode 100644 index 0000000000..5e8f4bd5b4 --- /dev/null +++ b/src/server/clickhouse/tracker.ts @@ -0,0 +1,641 @@ +// App-side ClickHouse Tracker (request/session/schema-coupled). Split out of the +// @civitai/clickhouse package, which keeps only the base client. Uses the singleton +// from the shim. +import { clickhouse } from '~/server/clickhouse/client'; +import type { NextApiRequest, NextApiResponse } from 'next'; +import type { Session } from 'next-auth'; +import requestIp from 'request-ip'; +import { isProd } from '~/env/other'; +import { env } from '~/env/server'; +import type { NewOrderImageRatingStatus, NsfwLevel } from '~/server/common/enums'; +import type { AllModKeys } from '~/server/jobs/entity-moderation'; +import { logToAxiom } from '~/server/logging/client'; +import { sleep } from '~/utils/errorHandling'; +import type { AddImageRatingInput } from '~/server/schema/games/new-order.schema'; +import type { ProhibitedSources } from '~/server/schema/user.schema'; +import type { NsfwLevelDeprecated } from '~/shared/constants/browsingLevel.constants'; +import dayjs from '~/shared/utils/dayjs'; +import type { + ArticleEngagementType, + BountyEngagementType, + EntityMetric_EntityType_Type, + EntityMetric_MetricType_Type, + EntityType, + NewOrderRankType, + ReportReason, + ReportStatus, + ReviewReactions, +} from '~/shared/utils/prisma/enums'; +import { createLogger } from '~/utils/logging'; +import { getServerAuthSession } from '~/server/auth/get-server-auth-session'; + +export type ViewType = + | 'ProfileView' + | 'ImageView' + | 'PostView' + | 'ModelView' + | 'ModelVersionView' + | 'ArticleView' + | 'CollectionView' + | 'BountyView' + | 'BountyEntryView'; + +export type UserActivityType = + | 'Registration' + | 'Login' + | 'Account closure' + | 'Subscribe' + | 'Cancel' + | 'Donate' + | 'Adjust Moderated Content Settings' + | 'Banned' + | 'Unbanned' + | 'Muted' + | 'Unmuted' + | 'RemoveContent' + | 'ExcludedFromLeaderboard' + | 'UnexcludedFromLeaderboard'; +export type ModelVersionActivty = 'Create' | 'Publish' | 'Download' | 'Unpublish' | 'HideDownload'; +export type ModelActivty = + | 'Create' + | 'Publish' + | 'Update' + | 'Unpublish' + | 'Archive' + | 'Takedown' + | 'Delete' + | 'PermanentDelete' + | 'Transfer'; +export type ResourceReviewType = 'Create' | 'Delete' | 'Exclude' | 'Include' | 'Update'; +export type ReactionType = + | 'Images_Create' + | 'Images_Delete' + | 'Comment_Create' + | 'Comment_Delete' + | 'Review_Create' + | 'Review_Delete' + | 'Question_Create' + | 'Question_Delete' + | 'Answer_Create' + | 'Answer_Delete' + | 'BountyEntry_Create' + | 'BountyEntry_Delete' + | 'Article_Create' + | 'Article_Delete'; +export type ReportType = 'Create' | 'StatusChange'; +export type ModelEngagementType = 'Hide' | 'Favorite' | 'Delete' | 'Notify'; +export type TagEngagementType = 'Hide' | 'Allow'; +export type UserEngagementType = 'Follow' | 'Hide' | 'Delete'; +export type CommentType = + | 'Model' + | 'Image' + | 'Post' + | 'Comment' + | 'Review' + | 'Bounty' + | 'BountyEntry'; +export type CommentActivity = 'Create' | 'Delete' | 'Update' | 'Hide' | 'Unhide'; +export type PostActivityType = 'Create' | 'Publish' | 'Tags' | 'Delete'; +export type ImageActivityType = + | 'Create' + | 'Delete' + | 'DeleteTOS' + | 'Tags' + | 'Resources' + | 'Restore'; +export type QuestionType = 'Create' | 'Delete'; +export type AnswerType = 'Create' | 'Delete'; +export type PartnerActivity = 'Run' | 'Update'; +export type BountyActivity = 'Create' | 'Update' | 'Delete' | 'Expire' | 'Refund'; +export type BountyEntryActivity = 'Create' | 'Update' | 'Delete' | 'Award'; +export type BountyBenefactorActivity = 'Create'; + +export type FileActivity = 'Download'; +export type ModelFileActivity = 'Create' | 'Delete' | 'Update'; + +export const ActionType = [ + 'AddToBounty_Click', + 'AddToBounty_Confirm', + 'AwardBounty_Click', + 'AwardBounty_Confirm', + 'Tip_Click', + 'Tip_Confirm', + 'TipInteractive_Click', + 'TipInteractive_Cancel', + 'NotEnoughFunds', + 'PurchaseFunds_Cancel', + 'PurchaseFunds_Confirm', + 'LoginRedirect', + 'Membership_Cancel', + 'Membership_Downgrade', + 'CSAM_Help_Triggered', + 'ProfanitySearch', + 'BuzzLimit_Set', + // Generation funnel telemetry — top-of-funnel clicks + form submission. + // Joined to orchestration.jobs / images_created downstream by userId + ts. + 'Model_Create_Click', + 'Image_Remix_Click', + 'Generator_Submit', +] as const; +export type ActionType = (typeof ActionType)[number]; + +export type TrackRequest = { + userId: number; + ip: string; + userAgent: string; +}; + +/** Track a webhook event to ClickHouse (fire and forget) */ +export async function trackWebhookEvent(type: string, payload: string) { + if (!clickhouse) return; + + try { + await clickhouse.insert({ + table: 'webhook_events_buffer', + values: [{ type, payload }], + format: 'JSONEachRow', + }); + } catch (error: any) { + console.error(`Failed to track ${type} webhook to ClickHouse:`, error.message); + } +} + +export class Tracker { + private actor: TrackRequest = { + userId: 0, + ip: 'unknown', + userAgent: 'unknown', + }; + private session: Session | null = null; + private req: NextApiRequest | undefined; + private res: NextApiResponse | undefined; + + private async resolveSession() { + if (!this.session && this.req && this.res) { + try { + await getServerAuthSession({ req: this.req, res: this.res }).then((session) => { + this.session = session; + this.actor.userId = session?.user?.id ?? this.actor.userId; + return session; + }); + } catch (e) { + const error = e as Error; + logToAxiom( + { + type: 'error', + name: 'Failed session', + message: error.message, + stack: error.stack, + cause: error.cause, + }, + 'clickhouse' + ); + } + } + } + + constructor(req?: NextApiRequest, res?: NextApiResponse) { + if (req && res) { + this.req = req; + this.res = res; + this.actor.ip = requestIp.getClientIp(req) ?? this.actor.ip; + this.actor.userAgent = req.headers['user-agent'] ?? this.actor.userAgent; + } + } + + private async send( + table: string, + data: object | ((args: { session: Session | null; actor: TrackRequest }) => object) + ) { + if (!env.CLICKHOUSE_TRACKER_URL) return; + await this.resolveSession(); + + const body = + typeof data === 'function' ? data({ session: this.session, actor: this.actor }) : data; + const url = `${env.CLICKHOUSE_TRACKER_URL}/track/${table}`; + + // Fire-and-forget at the call site, but the inner attempt loop checks + // HTTP status (not just network errors) and retries 5xx with backoff. + // Prior version only handled network rejection from fetch(), so any + // 5xx response from the tracker — common when NATS publish ack times + // out — was silently dropped. + void this.sendWithRetry(url, body, table); + } + + private async sendWithRetry( + url: string, + body: object, + table: string, + attempt = 1 + ): Promise { + const MAX_ATTEMPTS = 3; + const baseDelayMs = 250; + + try { + const res = await fetch(url, { + method: 'POST', + body: JSON.stringify(body), + headers: { 'Content-Type': 'application/json' }, + }); + + if (res.ok) return; + + // 4xx: tracker rejected the payload. Retrying won't help. Log and bail. + if (res.status >= 400 && res.status < 500) { + const errBody = await res.text().catch(() => ''); + logToAxiom( + { + type: 'warning', + name: 'Failed to track (4xx)', + details: { table, status: res.status, attempt, response: errBody.slice(0, 500) }, + message: `Tracker returned ${res.status}`, + }, + 'clickhouse' + ).catch(() => {}); + return; + } + + // 5xx: transient — NATS publish timeout, JetStream rejection, etc. + if (attempt < MAX_ATTEMPTS) { + await sleep(baseDelayMs * 2 ** (attempt - 1) + Math.random() * baseDelayMs); + return this.sendWithRetry(url, body, table, attempt + 1); + } + + const errBody = await res.text().catch(() => ''); + logToAxiom( + { + type: 'error', + name: 'Failed to track (5xx, exhausted)', + details: { table, status: res.status, attempts: attempt, response: errBody.slice(0, 500) }, + message: `Tracker returned ${res.status} after ${attempt} attempts`, + }, + 'clickhouse' + ).catch(() => {}); + } catch (e) { + const error = e as Error; + // Network-level failure. Retry the same as 5xx. + if (attempt < MAX_ATTEMPTS) { + await sleep(baseDelayMs * 2 ** (attempt - 1) + Math.random() * baseDelayMs); + return this.sendWithRetry(url, body, table, attempt + 1); + } + logToAxiom( + { + type: 'error', + name: 'Failed to track (network, exhausted)', + details: { table, attempts: attempt }, + message: error.message, + stack: error.stack, + cause: error.cause, + }, + 'clickhouse' + ).catch(() => {}); + } + } + + private async sendMany( + table: string, + data: object[] | ((args: { session: Session | null; actor: TrackRequest }) => object[]) + ) { + if (!clickhouse) return; + await this.resolveSession(); + const values = + typeof data === 'function' ? data({ session: this.session, actor: this.actor }) : data; + + try { + await clickhouse.insert({ + table, + values, + format: 'JSONEachRow', + }); + } catch (e) { + const error = e as Error; + logToAxiom( + { + type: 'error', + name: 'Failed to track', + details: { table, data: JSON.stringify(data) }, + message: error.message, + stack: error.stack, + cause: error.cause, + }, + 'clickhouse' + ).catch(); + } + } + + private async track( + table: string, + custom: object | ((session: Session | null) => object), + options?: { skipActorMeta: boolean } + ): Promise { + const { skipActorMeta = false } = options ?? {}; + + await this.send(table, ({ session, actor }) => { + const actorMeta = skipActorMeta ? { userId: actor.userId } : { ...actor }; + const customData = typeof custom === 'function' ? custom(session) : custom; + + return { + ...actorMeta, + ...customData, + }; + }); + } + + private async trackMany( + table: string, + custom: object[] | ((session: Session | null) => object[]), + options?: { skipActorMeta: boolean } + ) { + const { skipActorMeta = false } = options ?? {}; + + await this.sendMany(table, ({ session, actor }) => { + const actorMeta = skipActorMeta ? { userId: actor.userId } : { ...actor }; + const customData = typeof custom === 'function' ? custom(session) : custom; + return customData.map((custom) => ({ + ...actorMeta, + ...custom, + })); + }); + } + + public view(values: { type: ViewType; entityType: EntityType; entityId: number }) { + return this.track('views', values); + } + + public pageView(values: { + pageId: string; + path: string; + host: string; + ads: boolean; + country: string; + duration: number; + windowWidth: number; + windowHeight: number; + }) { + return this.send('pageViews', ({ session, actor }) => { + return { + userId: actor.userId, + memberType: session?.user?.tier ?? 'undefined', + ip: actor.ip, + ...values, + }; + }); + } + + public action(values: { type: ActionType; details?: any }) { + const { details, ...rest } = values; + return this.track('actions', { + ...rest, + details: + details != null ? (typeof details === 'string' ? details : JSON.stringify(details)) : '', + }); + } + + public activity(activity: string) { + return this.track('activities', { activity }); + } + + public bugReport(values: { bugId: number; status: string }) { + return this.track('bugReports', values); + } + + public modelEvent(values: { type: ModelActivty; modelId: number; nsfw: boolean }) { + return this.track('modelEvents', values); + } + + public redeemableCode(activity: string, details: { quantity?: number; code?: string }) { + return this.track('redeemableCodes', { activity, ...details }); + } + + public modelVersionEvent(values: { + type: ModelVersionActivty; + modelId: number; + modelVersionId: number; + nsfw: boolean; + earlyAccess?: boolean; + time?: Date; + fileId?: number; + }) { + return this.track('modelVersionEvents', values); + } + + public partnerEvent(values: { + type: PartnerActivity; + partnerId: number; + modelId?: number; + modelVersionId?: number; + nsfw?: boolean; + }) { + return this.track('partnerEvents', values); + } + + public userActivity(values: { + type: UserActivityType; + targetUserId: number; + source?: string; + landingPage?: string; + }) { + return this.track('userActivities', values); + } + + public resourceReview(values: { + type: ResourceReviewType; + modelId: number; + modelVersionId: number; + nsfw: boolean; + rating: number; + }) { + return this.track('resourceReviews', values); + } + + public reaction(values: { + type: ReactionType; + entityId: number; + ownerId: number; + reaction: ReviewReactions; + nsfw: NsfwLevelDeprecated; + }) { + return this.track('reactions', values); + } + + public question(values: { type: QuestionType; questionId: number }) { + return this.track('questions', values); + } + + public answer(values: { type: AnswerType; questionId: number; answerId: number }) { + return this.track('answers', values); + } + + public comment(values: { type: CommentType; entityId: number; nsfw: boolean }) { + return this.track('comments', values); + } + + public commentEvent(values: { type: CommentActivity; commentId: number }) { + return this.track('commentEvents', values); + } + + public post(values: { type: PostActivityType; postId: number; nsfw: boolean; tags: string[] }) { + return this.track('posts', values); + } + + public modelFile(values: { type: ModelFileActivity; id: number; modelVersionId: number }) { + return this.track('modelFileEvents', values); + } + + public images( + values: { + type: ImageActivityType; + imageId: number; + nsfw: NsfwLevelDeprecated; + tags: string[]; + ownerId: number; + tosReason?: string; + violationType?: string; + violationDetails?: string; + resources?: number[]; + userId?: number; + }[] + ) { + return this.trackMany('images', values); + } + + public bounty(values: { type: BountyActivity; bountyId: number; userId?: number }) { + return this.track('bounties', values); + } + + public bountyEntry(values: { + type: BountyEntryActivity; + bountyEntryId: number; + benefactorId?: number; + userId?: number; + }) { + return this.track('bountyEntries', values); + } + + public bountyBenefactor(values: { + type: BountyBenefactorActivity; + bountyId: number; + userId: number; + }) { + return this.track('bountyBenefactors', values); + } + + public modelEngagement(values: { type: ModelEngagementType; modelId: number }) { + return this.track('modelEngagements', values); + } + + public articleEngagement(values: { + type: ArticleEngagementType | `Delete${ArticleEngagementType}`; + articleId: number; + }) { + return this.track('articleEngagements', values); + } + + public tagEngagement(values: { type: TagEngagementType; tagId: number }) { + return this.track('tagEngagements', values); + } + + public userEngagement(values: { type: UserEngagementType; targetUserId: number }) { + return this.track('userEngagements', values); + } + + public bountyEngagement(values: { + type: BountyEngagementType | `Delete${BountyEngagementType}`; + bountyId: number; + }) { + return this.track('bountyEngagements', values); + } + + public prohibitedRequest(values: { + prompt: string; + negativePrompt: string; + source?: ProhibitedSources; + remixOfId?: number; + }) { + return this.track('prohibitedRequests', values); + } + + public report(values: { + type: ReportType; + entityType: string; + entityId: number; + reason: ReportReason; + status: ReportStatus; + }) { + return this.track('reports', values); + } + + public share(values: { url: string; platform: 'reddit' | 'twitter' | 'clipboard' }) { + return this.track('shares', values); + } + + public file(values: { type: FileActivity; entityType: string; entityId: number }) { + return this.track('files', values); + } + + public search(values: { query: string; index: string; filters?: any }) { + const { filters, ...rest } = values; + return this.track('search', { + ...rest, + filters: + filters != null ? (typeof filters === 'string' ? filters : JSON.stringify(filters)) : '', + }); + } + + public newOrderImageRating( + values: AddImageRatingInput & { + userId: number; + status: NewOrderImageRatingStatus; + grantedExp: number; + multiplier: number; + rank: NewOrderRankType; + originalLevel?: NsfwLevel; + voteWeight?: number; + } + ) { + return this.track('knights_new_order_image_rating', { ...values, createdAt: new Date() }); + } + + public entityMetric(values: { + entityType: EntityMetric_EntityType_Type; + entityId: number; + metricType: EntityMetric_MetricType_Type; + metricValue: number; + }) { + return this.track( + 'entityMetricEvents', + { ...values, createdAt: new Date() }, + { skipActorMeta: true } + ); + } + + public moderationRequest(values: { + entityType: AllModKeys; + entityId: number; + userId: number; + rules: string[]; + // value: string; + date: Date; + valid?: boolean; + }) { + return this.track('moderationRequest', { ...values }, { skipActorMeta: true }); + } + + public retoolAudit(values: { + action: string; + privileged: boolean; + outcome: 'ok' | 'error'; + errorMsg?: string; + payload: Record; + affected?: Record; + }) { + return this.track('retoolAuditLog', { + action: values.action, + privileged: values.privileged ? 1 : 0, + outcome: values.outcome, + errorMsg: values.errorMsg ?? '', + payload: JSON.stringify(values.payload), + affected: values.affected ? JSON.stringify(values.affected) : '', + }); + } +} diff --git a/src/server/db/client.ts b/src/server/db/client.ts new file mode 100644 index 0000000000..8a3e531cc5 --- /dev/null +++ b/src/server/db/client.ts @@ -0,0 +1,28 @@ +// App shim for @civitai/db Prisma clients. The package owns the env schema + factory; +// the app injects the slow-query sink (→ Axiom), owns the HMR singleton + Next build +// guard, and re-exports dbRead/dbWrite for existing call sites. +import { createPrismaClients, type PrismaClients } from '@civitai/db/client'; +import { isProd } from '~/env/other'; +import { env } from '~/env/server'; +import { logToAxiom } from '~/server/logging/client'; + +export * from '@civitai/db/client'; + +declare global { + // eslint-disable-next-line no-var, vars-on-top + var __civitaiPrismaClients: PrismaClients | undefined; +} + +const make = (): PrismaClients => + createPrismaClients({ + onSlowQuery: ({ query, duration, target }) => logToAxiom({ query, duration, target }, 'db-logs'), + }); + +const clients: PrismaClients = env.IS_BUILD + ? { dbRead: undefined as never, dbWrite: undefined as never } + : isProd + ? make() + : (global.__civitaiPrismaClients ??= make()); + +export const dbRead = clients.dbRead; +export const dbWrite = clients.dbWrite; diff --git a/packages/civitai-db/src/datapacketDb.ts b/src/server/db/datapacketDb.ts similarity index 51% rename from packages/civitai-db/src/datapacketDb.ts rename to src/server/db/datapacketDb.ts index 0ca1f9a9ce..0ca8230511 100644 --- a/packages/civitai-db/src/datapacketDb.ts +++ b/src/server/db/datapacketDb.ts @@ -1,23 +1,20 @@ -import { types } from 'pg'; +// App shim: datapacket read pool. See pgDb.ts for the pattern. +import { getClient, type AugmentedPool } from '@civitai/db/db-helpers'; import { isProd } from '~/env/other'; -import type { AugmentedPool } from '~/server/db/db-helpers'; -import { getClient } from '~/server/db/db-helpers'; +import { createLogger } from '~/utils/logging'; + +const log = createLogger('pgDb', 'blue'); declare global { // eslint-disable-next-line no-var, vars-on-top var globalDatapacketDbRead: AugmentedPool | undefined; } -// Fix Dates -types.setTypeParser(types.builtins.TIMESTAMP, function (stringValue) { - return new Date(stringValue.replace(' ', 'T') + 'Z'); -}); - export let datapacketDbRead: AugmentedPool; if (isProd) { - datapacketDbRead = getClient({ instance: 'datapacketRead' }); + datapacketDbRead = getClient({ instance: 'datapacketRead', log }); } else { if (!global.globalDatapacketDbRead) - global.globalDatapacketDbRead = getClient({ instance: 'datapacketRead' }); + global.globalDatapacketDbRead = getClient({ instance: 'datapacketRead', log }); datapacketDbRead = global.globalDatapacketDbRead; } diff --git a/src/server/db/db-helpers.ts b/src/server/db/db-helpers.ts new file mode 100644 index 0000000000..441e11471c --- /dev/null +++ b/src/server/db/db-helpers.ts @@ -0,0 +1,15 @@ +// App shim for @civitai/db helpers. Re-exports the pure utils + pool factory from the +// package, and binds the app's dbWrite into the Prisma-dependent helpers so existing +// call sites (getCurrentLSN(), checkNotUpToDate(lsn), dbKV) keep their signatures. +export * from '@civitai/db/db-helpers'; + +import { + getCurrentLSN as _getCurrentLSN, + checkNotUpToDate as _checkNotUpToDate, + makeDbKV, +} from '@civitai/db/kv-helpers'; +import { dbWrite } from '~/server/db/client'; + +export const getCurrentLSN = () => _getCurrentLSN(dbWrite); +export const checkNotUpToDate = (lsn: string) => _checkNotUpToDate(dbWrite, lsn); +export const dbKV = makeDbKV(dbWrite); diff --git a/packages/civitai-db/src/notifDb.ts b/src/server/db/notifDb.ts similarity index 58% rename from packages/civitai-db/src/notifDb.ts rename to src/server/db/notifDb.ts index c96f6101c8..de8af5833f 100644 --- a/packages/civitai-db/src/notifDb.ts +++ b/src/server/db/notifDb.ts @@ -1,8 +1,10 @@ -import { types } from 'pg'; +// App shim: notification pg pools. See pgDb.ts for the pattern. +import { getClient, type AugmentedPool } from '@civitai/db/db-helpers'; import { isProd } from '~/env/other'; import { env } from '~/env/server'; -import type { AugmentedPool } from '~/server/db/db-helpers'; -import { getClient } from '~/server/db/db-helpers'; +import { createLogger } from '~/utils/logging'; + +const log = createLogger('pgDb', 'blue'); declare global { // eslint-disable-next-line no-var, vars-on-top @@ -11,23 +13,19 @@ declare global { var globalNotifWrite: AugmentedPool | undefined; } -// Fix Dates -types.setTypeParser(types.builtins.TIMESTAMP, function (stringValue) { - return new Date(stringValue.replace(' ', 'T') + 'Z'); -}); - export let notifDbWrite: AugmentedPool; export let notifDbRead: AugmentedPool; const singleClient = env.NOTIFICATION_DB_URL === env.NOTIFICATION_DB_REPLICA_URL; if (isProd) { - notifDbWrite = getClient({ instance: 'notification' }); - notifDbRead = singleClient ? notifDbWrite : getClient({ instance: 'notificationRead' }); + notifDbWrite = getClient({ instance: 'notification', log }); + notifDbRead = singleClient ? notifDbWrite : getClient({ instance: 'notificationRead', log }); } else { - if (!global.globalNotifWrite) global.globalNotifWrite = getClient({ instance: 'notification' }); + if (!global.globalNotifWrite) + global.globalNotifWrite = getClient({ instance: 'notification', log }); if (!global.globalNotifRead) global.globalNotifRead = singleClient ? global.globalNotifWrite - : getClient({ instance: 'notificationRead' }); + : getClient({ instance: 'notificationRead', log }); notifDbWrite = global.globalNotifWrite; notifDbRead = global.globalNotifRead; } diff --git a/packages/civitai-db/src/pgDb.ts b/src/server/db/pgDb.ts similarity index 64% rename from packages/civitai-db/src/pgDb.ts rename to src/server/db/pgDb.ts index 1ce3359811..cb3b3b31a7 100644 --- a/packages/civitai-db/src/pgDb.ts +++ b/src/server/db/pgDb.ts @@ -1,8 +1,12 @@ -import { types } from 'pg'; +// App shim: primary pg pools. Calls the package getClient factory (which owns env + +// the TIMESTAMP type parser), injects the debug logger, and owns the HMR globals + +// Next build guard. Re-exports the pool instances for existing call sites. +import { getClient, type AugmentedPool } from '@civitai/db/db-helpers'; import { isProd } from '~/env/other'; import { env } from '~/env/server'; -import type { AugmentedPool } from '~/server/db/db-helpers'; -import { getClient } from '~/server/db/db-helpers'; +import { createLogger } from '~/utils/logging'; + +const log = createLogger('pgDb', 'blue'); declare global { // eslint-disable-next-line no-var, vars-on-top @@ -13,11 +17,6 @@ declare global { var globalPgWrite: AugmentedPool | undefined; } -// Fix Dates -types.setTypeParser(types.builtins.TIMESTAMP, function (stringValue) { - return new Date(stringValue.replace(' ', 'T') + 'Z'); -}); - export let pgDbWrite: AugmentedPool; export let pgDbRead: AugmentedPool; export let pgDbReadLong: AugmentedPool; @@ -25,19 +24,19 @@ export let pgDbReadLong: AugmentedPool; if (!env.IS_BUILD) { const singleClient = env.DATABASE_REPLICA_URL === env.DATABASE_URL; if (isProd) { - pgDbWrite = getClient(); - pgDbRead = singleClient ? pgDbWrite : getClient({ instance: 'primaryRead' }); - pgDbReadLong = singleClient ? pgDbWrite : getClient({ instance: 'primaryReadLong' }); + pgDbWrite = getClient({ log }); + pgDbRead = singleClient ? pgDbWrite : getClient({ instance: 'primaryRead', log }); + pgDbReadLong = singleClient ? pgDbWrite : getClient({ instance: 'primaryReadLong', log }); } else { - if (!global.globalPgWrite) global.globalPgWrite = getClient(); + if (!global.globalPgWrite) global.globalPgWrite = getClient({ log }); if (!global.globalPgRead) global.globalPgRead = singleClient ? global.globalPgWrite - : getClient({ instance: 'primaryRead' }); + : getClient({ instance: 'primaryRead', log }); if (!global.globalPgReadLong) global.globalPgReadLong = singleClient ? global.globalPgWrite - : getClient({ instance: 'primaryReadLong' }); + : getClient({ instance: 'primaryReadLong', log }); pgDbWrite = global.globalPgWrite; pgDbRead = global.globalPgRead; pgDbReadLong = global.globalPgReadLong; diff --git a/src/server/logging/client.ts b/src/server/logging/client.ts new file mode 100644 index 0000000000..66b28338da --- /dev/null +++ b/src/server/logging/client.ts @@ -0,0 +1,12 @@ +// App shim for @civitai/axiom. The package owns its env schema (incl. PODNAME and +// LOG_ERRORS_TO_STDOUT), so the app just instantiates the logger and re-exports the +// names existing call sites import from '~/server/logging/client'. +import { createAxiomLogger, safeError } from '@civitai/axiom/client'; +import { env } from '~/env/server'; + +// The build guard is a Next.js concern, so it lives here in the app shim — not in +// the app-agnostic @civitai/axiom package. Skip the client during `next build`. +const noopLog = async (_data: MixedObject, _datastream?: string) => {}; + +export const logToAxiom = env.IS_BUILD ? noopLog : createAxiomLogger().logToAxiom; +export { safeError }; diff --git a/src/server/prom/client.ts b/src/server/prom/client.ts new file mode 100644 index 0000000000..70175ca649 --- /dev/null +++ b/src/server/prom/client.ts @@ -0,0 +1,106 @@ +// App shim for @civitai/telemetry. Re-exports the generic prom helpers + metric +// definitions, and registers the DB pool-depth gauges here — they compose the db +// pools + prom helpers, which is app-level glue, not infrastructure. +import client from 'prom-client'; +import { datapacketDbRead } from '~/server/db/datapacketDb'; +import { notifDbRead, notifDbWrite } from '~/server/db/notifDb'; +import { pgDbRead, pgDbReadLong, pgDbWrite } from '~/server/db/pgDb'; + +export * from '@civitai/telemetry/client'; + +// pgPoolAcquireHistogram is registered in @civitai/db's db-helpers, not here, to avoid +// a module-init cycle (this module imports pgDb → db-helpers, which would import the +// histogram back), which webpack's CJS chunking can break with a TDZ error at runtime. + +declare global { + // eslint-disable-next-line no-var, vars-on-top + var pgGaugeInitialized: boolean; +} + +if (!global.pgGaugeInitialized) { + new client.Gauge({ + name: 'node_postgres_read_total_count', + help: 'node postgres read total count', + collect() { + this.set(pgDbRead.totalCount); + }, + }); + new client.Gauge({ + name: 'node_postgres_read_idle_count', + help: 'node postgres read idle count', + collect() { + this.set(pgDbRead.idleCount); + }, + }); + new client.Gauge({ + name: 'node_postgres_read_waiting_count', + help: 'node postgres read waiting count', + collect() { + this.set(pgDbRead.waitingCount); + }, + }); + new client.Gauge({ + name: 'node_postgres_write_total_count', + help: 'node postgres write total count', + collect() { + this.set(pgDbWrite.totalCount); + }, + }); + new client.Gauge({ + name: 'node_postgres_write_idle_count', + help: 'node postgres write idle count', + collect() { + this.set(pgDbWrite.idleCount); + }, + }); + new client.Gauge({ + name: 'node_postgres_write_waiting_count', + help: 'node postgres write waiting count', + collect() { + this.set(pgDbWrite.waitingCount); + }, + }); + + // Labeled pool metrics for all pools + new client.Gauge({ + name: 'node_postgres_pool_total_count', + help: 'Total connections in pg pool', + labelNames: ['pool'], + collect() { + this.set({ pool: 'read' }, pgDbRead?.totalCount ?? 0); + this.set({ pool: 'write' }, pgDbWrite?.totalCount ?? 0); + this.set({ pool: 'read_long' }, pgDbReadLong?.totalCount ?? 0); + this.set({ pool: 'notif_read' }, notifDbRead?.totalCount ?? 0); + this.set({ pool: 'notif_write' }, notifDbWrite?.totalCount ?? 0); + this.set({ pool: 'datapacket_read' }, datapacketDbRead?.totalCount ?? 0); + }, + }); + new client.Gauge({ + name: 'node_postgres_pool_idle_count', + help: 'Idle connections in pg pool', + labelNames: ['pool'], + collect() { + this.set({ pool: 'read' }, pgDbRead?.idleCount ?? 0); + this.set({ pool: 'write' }, pgDbWrite?.idleCount ?? 0); + this.set({ pool: 'read_long' }, pgDbReadLong?.idleCount ?? 0); + this.set({ pool: 'notif_read' }, notifDbRead?.idleCount ?? 0); + this.set({ pool: 'notif_write' }, notifDbWrite?.idleCount ?? 0); + this.set({ pool: 'datapacket_read' }, datapacketDbRead?.idleCount ?? 0); + }, + }); + new client.Gauge({ + name: 'node_postgres_pool_waiting_count', + help: 'Waiting connections in pg pool', + labelNames: ['pool'], + collect() { + this.set({ pool: 'read' }, pgDbRead?.waitingCount ?? 0); + this.set({ pool: 'write' }, pgDbWrite?.waitingCount ?? 0); + this.set({ pool: 'read_long' }, pgDbReadLong?.waitingCount ?? 0); + this.set({ pool: 'notif_read' }, notifDbRead?.waitingCount ?? 0); + this.set({ pool: 'notif_write' }, notifDbWrite?.waitingCount ?? 0); + this.set({ pool: 'datapacket_read' }, datapacketDbRead?.waitingCount ?? 0); + }, + }); + + global.pgGaugeInitialized = true; +} diff --git a/src/server/redis/client.ts b/src/server/redis/client.ts new file mode 100644 index 0000000000..3528bcc248 --- /dev/null +++ b/src/server/redis/client.ts @@ -0,0 +1,36 @@ +// App shim for @civitai/redis. The package owns its env schema + the typed client +// wrappers and key definitions; the app injects behavior (debug logger + the Flipt +// failover policy), owns the HMR singleton + Next build guard, and re-exports the +// names existing call sites import from '~/server/redis/client'. +import { createRedisClients, type RedisClients } from '@civitai/redis/client'; +import { isProd } from '~/env/other'; +import { env } from '~/env/server'; +import { FLIPT_FEATURE_FLAGS, isFlipt } from '~/server/flipt/client'; +import { createLogger } from '~/utils/logging'; + +// Re-export the key definitions, types, and factory for other consumers. +export * from '@civitai/redis/client'; + +declare global { + // eslint-disable-next-line no-var, vars-on-top + var __civitaiRedisClients: RedisClients | undefined; +} + +const log = createLogger('redis', 'green'); + +const make = (): RedisClients => + createRedisClients({ + log, + isEnhancedFailoverEnabled: (ctx) => + isFlipt(FLIPT_FEATURE_FLAGS.REDIS_CLUSTER_ENHANCED_FAILOVER, 'redis-cluster', ctx), + }); + +// Build guard is a Next.js concern → lives here, not in the package. +const clients: RedisClients = env.IS_BUILD + ? { redis: undefined as never, sysRedis: undefined as never } + : isProd + ? make() + : (global.__civitaiRedisClients ??= make()); + +export const redis = clients.redis; +export const sysRedis = clients.sysRedis; diff --git a/src/server/utils/otel-helpers.ts b/src/server/utils/otel-helpers.ts new file mode 100644 index 0000000000..1d3b45204b --- /dev/null +++ b/src/server/utils/otel-helpers.ts @@ -0,0 +1,3 @@ +// Re-export shim: this module moved to the @civitai/telemetry package. +// Existing call sites import from '~/server/utils/otel-helpers' unchanged. +export * from '@civitai/telemetry/otel-helpers'; diff --git a/src/shared/utils/prisma/enums.ts b/src/shared/utils/prisma/enums.ts new file mode 100644 index 0000000000..05481d6c66 --- /dev/null +++ b/src/shared/utils/prisma/enums.ts @@ -0,0 +1,3 @@ +// Re-export shim: this module moved to the @civitai/db-schema package. +// Existing call sites import from '~/shared/utils/prisma/enums' unchanged. +export * from '@civitai/db-schema/enums'; diff --git a/src/shared/utils/prisma/models.ts b/src/shared/utils/prisma/models.ts new file mode 100644 index 0000000000..8cd18f3079 --- /dev/null +++ b/src/shared/utils/prisma/models.ts @@ -0,0 +1,3 @@ +// Re-export shim: this module moved to the @civitai/db-schema package. +// Existing call sites import from '~/shared/utils/prisma/models' unchanged. +export * from '@civitai/db-schema/models'; diff --git a/tsconfig.json b/tsconfig.json index 17eee2de48..e43c55d1aa 100644 --- a/tsconfig.json +++ b/tsconfig.json @@ -19,7 +19,19 @@ "noUncheckedIndexedAccess": false, // TODO swap to true "baseUrl": "src", "paths": { - "~/*": ["./*"] + "~/*": ["./*"], + "@civitai/db-schema": ["../packages/civitai-db-schema/src/index"], + "@civitai/db-schema/*": ["../packages/civitai-db-schema/src/*"], + "@civitai/db": ["../packages/civitai-db/src/index"], + "@civitai/db/*": ["../packages/civitai-db/src/*"], + "@civitai/redis": ["../packages/civitai-redis/src/index"], + "@civitai/redis/*": ["../packages/civitai-redis/src/*"], + "@civitai/clickhouse": ["../packages/civitai-clickhouse/src/index"], + "@civitai/clickhouse/*": ["../packages/civitai-clickhouse/src/*"], + "@civitai/axiom": ["../packages/civitai-axiom/src/index"], + "@civitai/axiom/*": ["../packages/civitai-axiom/src/*"], + "@civitai/telemetry": ["../packages/civitai-telemetry/src/index"], + "@civitai/telemetry/*": ["../packages/civitai-telemetry/src/*"] }, "noErrorTruncation": true, "plugins": [ @@ -36,6 +48,7 @@ // "**/*.mjs", "scripts/local-dev/*.ts", "src", + "packages/*/src", "tests", ".next/types/**/*.ts" ],