mirror of
https://github.com/civitai/civitai.git
synced 2026-09-20 22:08:18 +08:00
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<Config> 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) <noreply@anthropic.com>
This commit is contained in:
Briant Diehl
@@ -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/
|
||||
|
||||
+8
-1
@@ -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,
|
||||
|
||||
+2
-1
@@ -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"
|
||||
|
||||
@@ -0,0 +1,7 @@
|
||||
{
|
||||
"name": "@civitai/axiom",
|
||||
"version": "0.0.0",
|
||||
"private": true,
|
||||
"main": "./src/index.ts",
|
||||
"types": "./src/index.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<void>;
|
||||
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<AxiomConfig>` 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<AxiomConfig> = {}): 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 };
|
||||
}
|
||||
|
||||
@@ -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<AxiomConfig> 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;
|
||||
@@ -0,0 +1 @@
|
||||
export * from './client';
|
||||
@@ -0,0 +1,7 @@
|
||||
{
|
||||
"name": "@civitai/clickhouse",
|
||||
"version": "0.0.0",
|
||||
"private": true,
|
||||
"main": "./src/index.ts",
|
||||
"types": "./src/index.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: <T extends object>(
|
||||
@@ -35,73 +13,12 @@ export type CustomClickHouseClient = ClickHouseClient & {
|
||||
$exec: (query: TemplateStringsArray | string, ...values: any[]) => Promise<void>;
|
||||
};
|
||||
|
||||
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 <T extends object>(
|
||||
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<T>();
|
||||
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<ClickhouseConfig> & {
|
||||
/** 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 <T extends object>(
|
||||
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<T>();
|
||||
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<void> {
|
||||
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<void> {
|
||||
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<string, unknown>;
|
||||
affected?: Record<string, unknown>;
|
||||
}) {
|
||||
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;
|
||||
}
|
||||
|
||||
@@ -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<ClickhouseConfig>.
|
||||
export const clickhouseEnv = {
|
||||
host: parsed.data.CLICKHOUSE_HOST,
|
||||
username: parsed.data.CLICKHOUSE_USERNAME,
|
||||
password: parsed.data.CLICKHOUSE_PASSWORD,
|
||||
isProd,
|
||||
};
|
||||
|
||||
export type ClickhouseConfig = typeof clickhouseEnv;
|
||||
@@ -0,0 +1 @@
|
||||
export * from './client';
|
||||
@@ -0,0 +1,7 @@
|
||||
{
|
||||
"name": "@civitai/db-schema",
|
||||
"version": "0.0.0",
|
||||
"private": true,
|
||||
"main": "./src/index.ts",
|
||||
"types": "./src/index.ts"
|
||||
}
|
||||
@@ -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 {
|
||||
|
||||
@@ -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';
|
||||
@@ -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:*"
|
||||
}
|
||||
}
|
||||
@@ -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<DbConfig> & {
|
||||
/** 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 };
|
||||
}
|
||||
|
||||
@@ -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<unknown>;
|
||||
type TaskGenerator = () => Task | null;
|
||||
|
||||
function isTaskGenerator(arg: any): arg is TaskGenerator {
|
||||
return typeof arg === 'function';
|
||||
}
|
||||
|
||||
type LimitConcurrencyOptions = {
|
||||
limit: number;
|
||||
betweenTasksFn?: () => Promise<void>;
|
||||
};
|
||||
|
||||
export function limitConcurrency(
|
||||
tasksOrGenerator: Task[] | TaskGenerator,
|
||||
options?: LimitConcurrencyOptions | number
|
||||
): Promise<void> {
|
||||
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<Task | null> => {
|
||||
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();
|
||||
});
|
||||
}
|
||||
@@ -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<ClientInstanceType, string> = {
|
||||
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<DbConfig> & {
|
||||
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<ClientInstanceType, string> = {
|
||||
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<T>(
|
||||
};
|
||||
}
|
||||
|
||||
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<void>)[];
|
||||
@@ -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 <T>(key: string, defaultValue?: T) {
|
||||
const stored = await dbWrite.keyValue.findUnique({ where: { key } });
|
||||
return stored ? (stored.value as T) : defaultValue;
|
||||
},
|
||||
set: async function <T>(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
|
||||
`);
|
||||
},
|
||||
};
|
||||
|
||||
@@ -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<DbConfig> 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;
|
||||
@@ -0,0 +1,4 @@
|
||||
export * from './env';
|
||||
export * from './client';
|
||||
export * from './db-helpers';
|
||||
export * from './kv-helpers';
|
||||
@@ -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 <T>(key: string, defaultValue?: T) {
|
||||
const stored = await dbWrite.keyValue.findUnique({ where: { key } });
|
||||
return stored ? (stored.value as T) : defaultValue;
|
||||
},
|
||||
set: async function <T>(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
|
||||
`);
|
||||
},
|
||||
};
|
||||
}
|
||||
@@ -0,0 +1,7 @@
|
||||
{
|
||||
"name": "@civitai/redis",
|
||||
"version": "0.0.0",
|
||||
"private": true,
|
||||
"main": "./src/index.ts",
|
||||
"types": "./src/index.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<string, string>) => Promise<boolean>;
|
||||
|
||||
// inlined — was slugit from ~/utils/string-helpers
|
||||
const slugit = (value: string) => slugify(value, { lower: true, strict: true });
|
||||
|
||||
export type RedisKeyStringsCache = Values<typeof REDIS_KEYS>;
|
||||
export type RedisKeyStringsSys = Values<typeof REDIS_SYS_KEYS>;
|
||||
@@ -176,14 +181,12 @@ interface CustomRedisClientCache extends CustomRedisClient<RedisKeyTemplateCache
|
||||
// eslint-disable-next-line @typescript-eslint/no-empty-interface
|
||||
interface CustomRedisClientSys extends CustomRedisClient<RedisKeyTemplateSys> {}
|
||||
|
||||
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<string, ReturnType<typeof setInterval>>();
|
||||
@@ -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<string, string> = { 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<RedisKeyTemplateSys>('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<RedisConfig> & {
|
||||
/** 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
|
||||
|
||||
@@ -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<RedisConfig> 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;
|
||||
@@ -0,0 +1 @@
|
||||
export * from './client';
|
||||
@@ -0,0 +1,7 @@
|
||||
{
|
||||
"name": "@civitai/telemetry",
|
||||
"version": "0.0.0",
|
||||
"private": true,
|
||||
"main": "./src/index.ts",
|
||||
"types": "./src/index.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.
|
||||
|
||||
@@ -0,0 +1,2 @@
|
||||
export * from './client';
|
||||
export * from './otel-helpers';
|
||||
Generated
+16
@@ -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':
|
||||
|
||||
@@ -0,0 +1,4 @@
|
||||
packages:
|
||||
- '.'
|
||||
- 'packages/*'
|
||||
- 'apps/*'
|
||||
@@ -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);
|
||||
|
||||
@@ -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';
|
||||
@@ -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<void> {
|
||||
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<void> {
|
||||
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<string, unknown>;
|
||||
affected?: Record<string, unknown>;
|
||||
}) {
|
||||
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) : '',
|
||||
});
|
||||
}
|
||||
}
|
||||
@@ -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;
|
||||
@@ -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;
|
||||
}
|
||||
@@ -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);
|
||||
@@ -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;
|
||||
}
|
||||
@@ -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;
|
||||
@@ -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 };
|
||||
@@ -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;
|
||||
}
|
||||
@@ -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;
|
||||
@@ -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';
|
||||
@@ -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';
|
||||
@@ -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';
|
||||
+14
-1
@@ -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"
|
||||
],
|
||||
|
||||
Reference in New Issue
Block a user