|
| 1 | +import { createHash } from "node:crypto"; |
| 2 | +import { promisify } from "node:util"; |
| 3 | +import { gunzip as gunzipCallback, gzip as gzipCallback } from "node:zlib"; |
| 4 | +import { CreateBackgroundWorkerRequestBody } from "@trigger.dev/core/v3"; |
| 5 | +import type { PrismaClientOrTransaction, WorkerDeployment } from "@trigger.dev/database"; |
| 6 | +import { z } from "zod"; |
| 7 | +import { env } from "~/env.server"; |
| 8 | +import type { AuthenticatedEnvironment } from "~/services/apiAuth.server"; |
| 9 | +import { logger } from "~/services/logger.server"; |
| 10 | +import { singleton } from "~/utils/singleton"; |
| 11 | +import { FEATURE_FLAG } from "./featureFlags"; |
| 12 | +import pRetry from "p-retry"; |
| 13 | +import { makeFlag } from "./featureFlags.server"; |
| 14 | +import { formatStorageUri, getObjectStoreClient, parseStorageUri } from "./objectStore.server"; |
| 15 | +import { ObjectStoreClient, type ObjectStoreClientConfig } from "./objectStoreClient.server"; |
| 16 | + |
| 17 | +const gzip = promisify(gzipCallback); |
| 18 | +const gunzip = promisify(gunzipCallback); |
| 19 | + |
| 20 | +export const DEPLOYMENT_INDEX_FORMAT_VERSION = 1; |
| 21 | + |
| 22 | +/** |
| 23 | + * The full indexing payload of one deployment, as registration parsed it, plus |
| 24 | + * the identity it was captured under. Versions, runtime and every declared |
| 25 | + * resource live in `request.metadata`. |
| 26 | + */ |
| 27 | +export const DeploymentIndexEnvelope = z.object({ |
| 28 | + formatVersion: z.literal(DEPLOYMENT_INDEX_FORMAT_VERSION), |
| 29 | + projectId: z.string(), |
| 30 | + projectRef: z.string(), |
| 31 | + environmentId: z.string(), |
| 32 | + environmentSlug: z.string(), |
| 33 | + environmentType: z.string(), |
| 34 | + deploymentId: z.string(), |
| 35 | + deploymentVersion: z.string(), |
| 36 | + request: CreateBackgroundWorkerRequestBody, |
| 37 | +}); |
| 38 | + |
| 39 | +export type DeploymentIndexEnvelope = z.infer<typeof DeploymentIndexEnvelope>; |
| 40 | + |
| 41 | +/** What a deployment row stores to find and verify its snapshot. */ |
| 42 | +export type DeploymentIndexReference = { |
| 43 | + indexObjectKey: string; |
| 44 | + indexSha256: string; |
| 45 | + indexFormatVersion: number; |
| 46 | +}; |
| 47 | + |
| 48 | +export class DeploymentIndexError extends Error { |
| 49 | + constructor(message: string) { |
| 50 | + super(message); |
| 51 | + this.name = "DeploymentIndexError"; |
| 52 | + } |
| 53 | +} |
| 54 | + |
| 55 | +const CONTENT_TYPE = "application/gzip"; |
| 56 | + |
| 57 | +type DeploymentIndexStoreOptions = { |
| 58 | + /** The logical key's first segment: the dedicated bucket, or the packet root. */ |
| 59 | + root: string; |
| 60 | + /** The object-store protocol new snapshots are written with, recorded on their reference. */ |
| 61 | + protocol?: string; |
| 62 | + /** The client for a protocol; a reference is read back through the protocol it records. */ |
| 63 | + clientFor: (protocol: string | undefined) => ObjectStoreClient | undefined; |
| 64 | + maxAttempts: number; |
| 65 | + minRetryTimeoutMs?: number; |
| 66 | +}; |
| 67 | + |
| 68 | +/** |
| 69 | + * Immutable snapshots, one object per deployment and content. The key ends in |
| 70 | + * the sha256 of the uncompressed JSON, so a retried upload of the same index |
| 71 | + * lands on the same key and a create-only write can never replace it. |
| 72 | + */ |
| 73 | +export class DeploymentIndexStore { |
| 74 | + private readonly prefix: string; |
| 75 | + |
| 76 | + constructor(private readonly options: DeploymentIndexStoreOptions) { |
| 77 | + this.prefix = `${options.root}/deployment-indexes`; |
| 78 | + } |
| 79 | + |
| 80 | + async put(envelope: DeploymentIndexEnvelope): Promise<DeploymentIndexReference> { |
| 81 | + const { protocol } = this.options; |
| 82 | + const client = this.#client(protocol); |
| 83 | + const json = JSON.stringify(envelope); |
| 84 | + const sha256 = createHash("sha256").update(json).digest("hex"); |
| 85 | + const key = [ |
| 86 | + this.prefix, |
| 87 | + envelope.projectRef, |
| 88 | + envelope.environmentSlug, |
| 89 | + envelope.deploymentId, |
| 90 | + `${sha256}.json.gz`, |
| 91 | + ].join("/"); |
| 92 | + const body = await gzip(json); |
| 93 | + |
| 94 | + const result = await pRetry(() => client.putObjectIfAbsent(key, body, CONTENT_TYPE), { |
| 95 | + retries: this.options.maxAttempts - 1, |
| 96 | + minTimeout: this.options.minRetryTimeoutMs ?? 250, |
| 97 | + maxTimeout: 2000, |
| 98 | + factor: 2, |
| 99 | + onFailedAttempt: (error) => { |
| 100 | + logger.warn("Deployment index upload failed, retrying", { |
| 101 | + key, |
| 102 | + attempt: error.attemptNumber, |
| 103 | + error: error.message, |
| 104 | + }); |
| 105 | + }, |
| 106 | + }); |
| 107 | + |
| 108 | + logger.debug("Stored deployment index", { key, result, bytes: body.byteLength }); |
| 109 | + |
| 110 | + return { |
| 111 | + indexObjectKey: formatStorageUri(key, protocol), |
| 112 | + indexSha256: sha256, |
| 113 | + indexFormatVersion: envelope.formatVersion, |
| 114 | + }; |
| 115 | + } |
| 116 | + |
| 117 | + async get( |
| 118 | + reference: DeploymentIndexReference, |
| 119 | + owner: { projectId: string } |
| 120 | + ): Promise<DeploymentIndexEnvelope> { |
| 121 | + const { protocol, path: key } = parseStorageUri(reference.indexObjectKey); |
| 122 | + |
| 123 | + if (!key.startsWith(`${this.prefix}/`)) { |
| 124 | + throw new DeploymentIndexError("Deployment index is not in this store"); |
| 125 | + } |
| 126 | + |
| 127 | + if (reference.indexFormatVersion !== DEPLOYMENT_INDEX_FORMAT_VERSION) { |
| 128 | + throw new DeploymentIndexError( |
| 129 | + `Unsupported deployment index format version ${reference.indexFormatVersion}` |
| 130 | + ); |
| 131 | + } |
| 132 | + |
| 133 | + const response = await this.#client(protocol).getObjectResponse(key); |
| 134 | + const json = (await gunzip(Buffer.from(await response.arrayBuffer()))).toString("utf8"); |
| 135 | + const sha256 = createHash("sha256").update(json).digest("hex"); |
| 136 | + |
| 137 | + if (sha256 !== reference.indexSha256) { |
| 138 | + throw new DeploymentIndexError("Deployment index checksum mismatch"); |
| 139 | + } |
| 140 | + |
| 141 | + const parsed = DeploymentIndexEnvelope.safeParse(JSON.parse(json)); |
| 142 | + |
| 143 | + if (!parsed.success) { |
| 144 | + throw new DeploymentIndexError("Deployment index does not match its format version"); |
| 145 | + } |
| 146 | + |
| 147 | + if (parsed.data.projectId !== owner.projectId) { |
| 148 | + throw new DeploymentIndexError("Deployment index belongs to a different project"); |
| 149 | + } |
| 150 | + |
| 151 | + return parsed.data; |
| 152 | + } |
| 153 | + |
| 154 | + #client(protocol: string | undefined): ObjectStoreClient { |
| 155 | + const client = this.options.clientFor(protocol); |
| 156 | + if (!client) { |
| 157 | + throw new DeploymentIndexError( |
| 158 | + `No object store is configured for protocol ${protocol ?? "default"}` |
| 159 | + ); |
| 160 | + } |
| 161 | + return client; |
| 162 | + } |
| 163 | +} |
| 164 | + |
| 165 | +/** Packet keys live under this root, so shared-store snapshots sit beside them. */ |
| 166 | +const SHARED_STORE_ROOT = "packets"; |
| 167 | + |
| 168 | +export type DeploymentIndexStoreConfig = { |
| 169 | + /** A dedicated store for snapshots (DEPLOYMENT_INDEX_OBJECT_STORE_*). */ |
| 170 | + dedicated?: ObjectStoreClientConfig & { bucket: string }; |
| 171 | + /** The protocol new packet uploads use, when snapshots share the packet store. */ |
| 172 | + sharedProtocol?: string; |
| 173 | + /** The packet store's client for a protocol. */ |
| 174 | + sharedClientFor: (protocol: string | undefined) => ObjectStoreClient | undefined; |
| 175 | + maxAttempts: number; |
| 176 | +}; |
| 177 | + |
| 178 | +/** |
| 179 | + * The dedicated store when one is configured, otherwise the packet store's |
| 180 | + * provider for new uploads, otherwise undefined (no snapshots). |
| 181 | + */ |
| 182 | +export function createDeploymentIndexStore( |
| 183 | + config: DeploymentIndexStoreConfig |
| 184 | +): DeploymentIndexStore | undefined { |
| 185 | + const { dedicated, sharedProtocol, sharedClientFor, maxAttempts } = config; |
| 186 | + |
| 187 | + if (dedicated) { |
| 188 | + const client = ObjectStoreClient.create(dedicated); |
| 189 | + return new DeploymentIndexStore({ |
| 190 | + root: dedicated.bucket, |
| 191 | + clientFor: () => client, |
| 192 | + maxAttempts, |
| 193 | + }); |
| 194 | + } |
| 195 | + |
| 196 | + if (!sharedClientFor(sharedProtocol)) { |
| 197 | + return undefined; |
| 198 | + } |
| 199 | + |
| 200 | + return new DeploymentIndexStore({ |
| 201 | + root: SHARED_STORE_ROOT, |
| 202 | + protocol: sharedProtocol, |
| 203 | + clientFor: sharedClientFor, |
| 204 | + maxAttempts, |
| 205 | + }); |
| 206 | +} |
| 207 | + |
| 208 | +/** The store configured by DEPLOYMENT_INDEX_OBJECT_STORE_* and OBJECT_STORE_*. */ |
| 209 | +export const getDeploymentIndexStore = (): DeploymentIndexStore | undefined => |
| 210 | + singleton("deploymentIndexStore", () => { |
| 211 | + const baseUrl = env.DEPLOYMENT_INDEX_OBJECT_STORE_BASE_URL; |
| 212 | + const bucket = env.DEPLOYMENT_INDEX_OBJECT_STORE_BUCKET; |
| 213 | + |
| 214 | + return createDeploymentIndexStore({ |
| 215 | + dedicated: |
| 216 | + baseUrl && bucket |
| 217 | + ? { |
| 218 | + baseUrl, |
| 219 | + bucket, |
| 220 | + accessKeyId: env.DEPLOYMENT_INDEX_OBJECT_STORE_ACCESS_KEY_ID || undefined, |
| 221 | + secretAccessKey: env.DEPLOYMENT_INDEX_OBJECT_STORE_SECRET_ACCESS_KEY || undefined, |
| 222 | + region: env.DEPLOYMENT_INDEX_OBJECT_STORE_REGION || undefined, |
| 223 | + service: env.DEPLOYMENT_INDEX_OBJECT_STORE_SERVICE || undefined, |
| 224 | + } |
| 225 | + : undefined, |
| 226 | + sharedProtocol: env.OBJECT_STORE_DEFAULT_PROTOCOL, |
| 227 | + sharedClientFor: getObjectStoreClient, |
| 228 | + maxAttempts: env.DEPLOYMENT_INDEX_UPLOAD_MAX_ATTEMPTS, |
| 229 | + }); |
| 230 | + }); |
| 231 | + |
| 232 | +/** |
| 233 | + * Store the deployment's index before registration mutates anything. Returns |
| 234 | + * undefined when no store is configured or capture is switched off, and throws |
| 235 | + * once the bounded retries are spent. |
| 236 | + */ |
| 237 | +export async function captureDeploymentIndex( |
| 238 | + environment: AuthenticatedEnvironment, |
| 239 | + deployment: Pick<WorkerDeployment, "friendlyId" | "version">, |
| 240 | + body: CreateBackgroundWorkerRequestBody, |
| 241 | + store: DeploymentIndexStore | undefined, |
| 242 | + prisma: PrismaClientOrTransaction |
| 243 | +): Promise<DeploymentIndexReference | undefined> { |
| 244 | + if (!store) { |
| 245 | + return undefined; |
| 246 | + } |
| 247 | + |
| 248 | + const enabled = await makeFlag(prisma)({ |
| 249 | + key: FEATURE_FLAG.deploymentIndexCaptureEnabled, |
| 250 | + defaultValue: env.DEPLOYMENT_INDEX_CAPTURE_ENABLED, |
| 251 | + }); |
| 252 | + |
| 253 | + if (!enabled) { |
| 254 | + return undefined; |
| 255 | + } |
| 256 | + |
| 257 | + return store.put({ |
| 258 | + formatVersion: DEPLOYMENT_INDEX_FORMAT_VERSION, |
| 259 | + projectId: environment.projectId, |
| 260 | + projectRef: environment.project.externalRef, |
| 261 | + environmentId: environment.id, |
| 262 | + environmentSlug: environment.slug, |
| 263 | + environmentType: environment.type, |
| 264 | + deploymentId: deployment.friendlyId, |
| 265 | + deploymentVersion: deployment.version, |
| 266 | + request: body, |
| 267 | + }); |
| 268 | +} |
0 commit comments