diff --git a/.env.example b/.env.example index 9df3a90..9dffc3a 100644 --- a/.env.example +++ b/.env.example @@ -9,6 +9,8 @@ PHONE_ENCRYPTION_KEY= PHONE_HMAC_KEY= VERIFICATION_CODE_HMAC_KEY= REFRESH_TOKEN_HMAC_KEY= +LEASE_TOKEN_HMAC_KEY= +LEASE_TOKEN_ENCRYPTION_KEY= JWT_SECRET= JWT_ISSUER=drift-api JWT_AUDIENCE=drift-web diff --git a/apps/api/package.json b/apps/api/package.json index 9de92b5..3df4876 100644 --- a/apps/api/package.json +++ b/apps/api/package.json @@ -7,7 +7,7 @@ "build": "tsc -p tsconfig.build.json", "start": "node dist/main.js", "test": "vitest run --config vitest.config.ts --no-file-parallelism", - "test:e2e": "vitest run --config vitest.config.ts src/health/health.e2e-spec.ts src/auth/auth.e2e-spec.ts src/profile/profile.e2e-spec.ts src/bottle/bottle.e2e-spec.ts --no-file-parallelism", + "test:e2e": "vitest run --config vitest.config.ts src/health/health.e2e-spec.ts src/auth/auth.e2e-spec.ts src/profile/profile.e2e-spec.ts src/bottle/bottle.e2e-spec.ts src/match/match.e2e-spec.ts --no-file-parallelism", "typecheck": "tsc --noEmit -p tsconfig.json" }, "dependencies": { diff --git a/apps/api/src/app.module.ts b/apps/api/src/app.module.ts index 62b9b6f..bbb4184 100644 --- a/apps/api/src/app.module.ts +++ b/apps/api/src/app.module.ts @@ -4,8 +4,11 @@ import { HealthModule } from "./health/health.module.js"; import { AuthModule } from "./auth/auth.module.js"; import { ProfileModule } from "./profile/profile.module.js"; import { BottleModule } from "./bottle/bottle.module.js"; +import { MatchModule } from "./match/match.module.js"; -@Module({ imports: [HealthModule, AuthModule, ProfileModule, BottleModule] }) +@Module({ + imports: [HealthModule, AuthModule, ProfileModule, BottleModule, MatchModule], +}) export class AppModule implements NestModule { configure(consumer: MiddlewareConsumer): void { consumer.apply(RequestIdMiddleware).forRoutes("{*path}"); diff --git a/apps/api/src/auth/auth.config.spec.ts b/apps/api/src/auth/auth.config.spec.ts index 84663f3..18f7484 100644 --- a/apps/api/src/auth/auth.config.spec.ts +++ b/apps/api/src/auth/auth.config.spec.ts @@ -1,7 +1,9 @@ import { afterEach, beforeEach, describe, expect, it } from "vitest"; import { codeHmac, + decryptLeaseToken, demoSmsCodeEnabled, + encryptLeaseToken, phoneHmac, resetAuthEnvironmentForTests, validateAuthEnvironment, @@ -13,6 +15,8 @@ const valid = { PHONE_HMAC_KEY: "phone-hmac-key-that-is-at-least-32-bytes", VERIFICATION_CODE_HMAC_KEY: "code-hmac-key-that-is-at-least-32-bytes!", REFRESH_TOKEN_HMAC_KEY: "refresh-key-that-is-at-least-thirty-two-bytes", + LEASE_TOKEN_HMAC_KEY: "lease-hmac-key-that-is-at-least-thirty-two-bytes", + LEASE_TOKEN_ENCRYPTION_KEY: "BBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBB=", JWT_SECRET: "jwt-secret-that-is-at-least-thirty-two-bytes", WEB_ORIGIN: "http://localhost:3000", }; @@ -46,6 +50,33 @@ describe("auth environment", () => { ); }); + it("rejects lease keys reused from any authentication secret", () => { + process.env.LEASE_TOKEN_HMAC_KEY = process.env.REFRESH_TOKEN_HMAC_KEY; + expect(() => validateAuthEnvironment()).toThrow( + /secrets must be independent/i, + ); + }); + + it("binds canonical lease token ciphertext to its lease id", () => { + validateAuthEnvironment(); + const leaseId = "00000000-0000-4000-8000-000000000001"; + const token = "x".repeat(43); + const ciphertext = encryptLeaseToken(token, leaseId); + expect(ciphertext).toHaveLength(71); + expect(decryptLeaseToken(ciphertext, leaseId)).toBe(token); + expect(() => + decryptLeaseToken(ciphertext, "00000000-0000-4000-8000-000000000002"), + ).toThrow(); + }); + + it("rejects encrypted lease plaintext with a non-canonical format", () => { + validateAuthEnvironment(); + const leaseId = "00000000-0000-4000-8000-000000000001"; + expect(() => + decryptLeaseToken(encryptLeaseToken("short", leaseId), leaseId), + ).toThrow(/Invalid lease token ciphertext/); + }); + it("uses a verification-code key independent from the phone key", () => { validateAuthEnvironment(); const phoneDigest = phoneHmac("+8613800138000"); diff --git a/apps/api/src/auth/auth.config.ts b/apps/api/src/auth/auth.config.ts index a44bd3b..50d1dcb 100644 --- a/apps/api/src/auth/auth.config.ts +++ b/apps/api/src/auth/auth.config.ts @@ -9,10 +9,9 @@ import { function required(name: string, min = 32): Buffer { const value = process.env[name]; if (!value) throw new Error(`${name} is required`); - const decoded = - name === "PHONE_ENCRYPTION_KEY" - ? Buffer.from(value, "base64") - : Buffer.from(value); + const decoded = name.endsWith("_ENCRYPTION_KEY") + ? Buffer.from(value, "base64") + : Buffer.from(value); if (decoded.length < min) throw new Error(`${name} must be at least ${min} bytes`); return decoded; @@ -22,6 +21,8 @@ const REQUIRED_KEYS = [ "PHONE_HMAC_KEY", "VERIFICATION_CODE_HMAC_KEY", "REFRESH_TOKEN_HMAC_KEY", + "LEASE_TOKEN_HMAC_KEY", + "LEASE_TOKEN_ENCRYPTION_KEY", "JWT_SECRET", ] as const; type AuthSecrets = Record<(typeof REQUIRED_KEYS)[number], Buffer>; @@ -56,8 +57,13 @@ export function validateAuthEnvironment(): void { const loaded = Object.fromEntries( REQUIRED_KEYS.map((name) => [name, required(name)]), ) as unknown as AuthSecrets; - if (loaded.PHONE_ENCRYPTION_KEY.length !== 32) - throw new Error("PHONE_ENCRYPTION_KEY must decode to exactly 32 bytes"); + for (const name of [ + "PHONE_ENCRYPTION_KEY", + "LEASE_TOKEN_ENCRYPTION_KEY", + ] as const) { + if (loaded[name].length !== 32) + throw new Error(`${name} must decode to exactly 32 bytes`); + } const fingerprints = REQUIRED_KEYS.map((name) => loaded[name].toString("hex"), ); @@ -103,6 +109,43 @@ export const refreshHmac = (token: string): string => createHmac("sha256", authSecrets().REFRESH_TOKEN_HMAC_KEY) .update(token) .digest("hex"); +export const leaseHmac = (token: string): string => + createHmac("sha256", authSecrets().LEASE_TOKEN_HMAC_KEY) + .update(token) + .digest("hex"); +export function encryptLeaseToken( + token: string, + leaseId: string, +): Uint8Array { + const nonce = randomBytes(12); + const cipher = createCipheriv( + "aes-256-gcm", + authSecrets().LEASE_TOKEN_ENCRYPTION_KEY, + nonce, + ); + cipher.setAAD(Buffer.from(leaseId)); + const body = Buffer.concat([cipher.update(token, "utf8"), cipher.final()]); + const encrypted = Buffer.concat([nonce, cipher.getAuthTag(), body]); + return new Uint8Array(encrypted).slice(); +} +export function decryptLeaseToken(value: Uint8Array, leaseId: string): string { + // nonce (12) + authentication tag (16) + canonical 32-byte token (43 base64url chars) + if (value.length !== 71) throw new Error("Invalid lease token ciphertext"); + const decipher = createDecipheriv( + "aes-256-gcm", + authSecrets().LEASE_TOKEN_ENCRYPTION_KEY, + Buffer.from(value.subarray(0, 12)), + ); + decipher.setAAD(Buffer.from(leaseId)); + decipher.setAuthTag(Buffer.from(value.subarray(12, 28))); + const token = Buffer.concat([ + decipher.update(Buffer.from(value.subarray(28))), + decipher.final(), + ]).toString(); + if (!/^[A-Za-z0-9_-]{43}$/.test(token)) + throw new Error("Invalid lease token ciphertext"); + return token; +} export const safeEqual = (a: string, b: string): boolean => { const x = Buffer.from(a); const y = Buffer.from(b); diff --git a/apps/api/src/auth/auth.e2e-spec.ts b/apps/api/src/auth/auth.e2e-spec.ts index e50bdb0..d299e29 100644 --- a/apps/api/src/auth/auth.e2e-spec.ts +++ b/apps/api/src/auth/auth.e2e-spec.ts @@ -30,6 +30,9 @@ process.env.PHONE_ENCRYPTION_KEY = process.env.PHONE_HMAC_KEY = "test-phone-hmac-key-with-at-least-32-bytes"; process.env.VERIFICATION_CODE_HMAC_KEY = "test-code-hmac-key-with-at-least-32-bytes"; +process.env.LEASE_TOKEN_HMAC_KEY = "test-lease-hmac-key-with-at-least-32-bytes"; +process.env.LEASE_TOKEN_ENCRYPTION_KEY = + "BBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBB="; process.env.JWT_SECRET = "test-jwt-secret-with-at-least-thirty-two-bytes"; process.env.REFRESH_TOKEN_HMAC_KEY = "test-refresh-hmac-key-with-at-least-32-bytes"; diff --git a/apps/api/src/auth/token.service.spec.ts b/apps/api/src/auth/token.service.spec.ts index 08a2d90..590d716 100644 --- a/apps/api/src/auth/token.service.spec.ts +++ b/apps/api/src/auth/token.service.spec.ts @@ -11,6 +11,9 @@ process.env.VERIFICATION_CODE_HMAC_KEY = "token-code-hmac-key-with-at-least-32-bytes!"; process.env.REFRESH_TOKEN_HMAC_KEY = "token-refresh-hmac-key-with-at-least-32-bytes"; +process.env.LEASE_TOKEN_HMAC_KEY = "test-lease-hmac-key-with-at-least-32-bytes"; +process.env.LEASE_TOKEN_ENCRYPTION_KEY = + "BBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBB="; process.env.JWT_SECRET = "token-jwt-secret-with-at-least-thirty-two-bytes"; const secret = process.env.JWT_SECRET; diff --git a/apps/api/src/bottle/bottle.e2e-spec.ts b/apps/api/src/bottle/bottle.e2e-spec.ts index 07ca4af..dafa779 100644 --- a/apps/api/src/bottle/bottle.e2e-spec.ts +++ b/apps/api/src/bottle/bottle.e2e-spec.ts @@ -19,6 +19,9 @@ process.env.PHONE_ENCRYPTION_KEY = process.env.PHONE_HMAC_KEY = "test-phone-hmac-key-with-at-least-32-bytes"; process.env.VERIFICATION_CODE_HMAC_KEY = "test-code-hmac-key-with-at-least-32-bytes"; +process.env.LEASE_TOKEN_HMAC_KEY = "test-lease-hmac-key-with-at-least-32-bytes"; +process.env.LEASE_TOKEN_ENCRYPTION_KEY = + "BBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBB="; process.env.JWT_SECRET = "test-jwt-secret-with-at-least-thirty-two-bytes"; process.env.REFRESH_TOKEN_HMAC_KEY = "test-refresh-hmac-key-with-at-least-32-bytes"; diff --git a/apps/api/src/health/health.e2e-spec.ts b/apps/api/src/health/health.e2e-spec.ts index f975902..6350663 100644 --- a/apps/api/src/health/health.e2e-spec.ts +++ b/apps/api/src/health/health.e2e-spec.ts @@ -25,6 +25,21 @@ import { PrismaService } from "../database/prisma.service.js"; import { bootstrap, configureApp } from "../main.js"; import { HealthService } from "./health.service.js"; +process.env.NODE_ENV = "test"; +process.env.WEB_ORIGIN = "http://localhost:3000"; +process.env.PHONE_ENCRYPTION_KEY = + "AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA="; +process.env.PHONE_HMAC_KEY = "health-phone-hmac-key-with-at-least-32-bytes"; +process.env.VERIFICATION_CODE_HMAC_KEY = + "health-code-hmac-key-with-at-least-32-bytes"; +process.env.REFRESH_TOKEN_HMAC_KEY = + "health-refresh-hmac-key-with-at-least-32-bytes"; +process.env.LEASE_TOKEN_HMAC_KEY = + "health-lease-hmac-key-with-at-least-32-bytes"; +process.env.LEASE_TOKEN_ENCRYPTION_KEY = + "BBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBB="; +process.env.JWT_SECRET = "health-jwt-secret-with-at-least-thirty-two-bytes"; + type RedisClientDouble = EventEmitter & { isOpen: boolean; connect: ReturnType; diff --git a/apps/api/src/match/candidate.repository.ts b/apps/api/src/match/candidate.repository.ts new file mode 100644 index 0000000..314bf0f --- /dev/null +++ b/apps/api/src/match/candidate.repository.ts @@ -0,0 +1,41 @@ +import { Injectable } from "@nestjs/common"; +import type { Prisma } from "@prisma/client"; + +export type Candidate = { + id: string; + version: number; + authorId: string; + contentText: string; + publicId: string; + nickname: string; + avatarColor: string; + bio: string | null; +}; + +@Injectable() +export class CandidateRepository { + async findBatch(tx: Prisma.TransactionClient, pickerId: string, limit = 32) { + return tx.$queryRaw` + SELECT b."id", b."version", b."author_id" AS "authorId", + b."content_text" AS "contentText", p."public_id" AS "publicId", + p."nickname", p."avatar_color" AS "avatarColor", p."bio" + FROM "bottles" b + JOIN "anonymous_profiles" p ON p."account_id" = b."author_id" + WHERE b."author_id" <> ${pickerId}::uuid + AND b."review_status" = 'APPROVED' + AND b."pool_status" = 'IN_POOL' + AND b."active_lease_id" IS NULL + AND p."review_status" = 'APPROVED' + AND NOT EXISTS ( + SELECT 1 FROM "bottle_pick_history" h + WHERE h."bottle_id" = b."id" AND h."picker_id" = ${pickerId}::uuid + ) + AND NOT EXISTS ( + SELECT 1 FROM "blocks" x + WHERE (x."blocker_id" = ${pickerId}::uuid AND x."blocked_id" = b."author_id") + OR (x."blocker_id" = b."author_id" AND x."blocked_id" = ${pickerId}::uuid) + ) + ORDER BY b."created_at", b."id" + LIMIT ${limit}`; + } +} diff --git a/apps/api/src/match/dto.ts b/apps/api/src/match/dto.ts new file mode 100644 index 0000000..e49301b --- /dev/null +++ b/apps/api/src/match/dto.ts @@ -0,0 +1,6 @@ +import { IsString, IsUUID, Matches } from "class-validator"; + +export class ReturnBottleDto { + @IsUUID() leaseId!: string; + @IsString() @Matches(/^[A-Za-z0-9_-]{43}$/) token!: string; +} diff --git a/apps/api/src/match/match.controller.ts b/apps/api/src/match/match.controller.ts new file mode 100644 index 0000000..3e6ff4d --- /dev/null +++ b/apps/api/src/match/match.controller.ts @@ -0,0 +1,58 @@ +import { + Body, + Controller, + Headers, + HttpStatus, + Inject, + Param, + ParseUUIDPipe, + Post, + UseGuards, +} from "@nestjs/common"; +import { ErrorCode } from "@drift/contracts"; +import { AuthGuard } from "../auth/auth.guard.js"; +import { CurrentUser } from "../auth/current-user.decorator.js"; +import type { AccessClaims } from "../auth/token.service.js"; +import { DomainException } from "../common/domain.exception.js"; +import { MatchService } from "./match.service.js"; +import { ReturnBottleDto } from "./dto.js"; + +@Controller() +@UseGuards(AuthGuard) +export class MatchController { + constructor(@Inject(MatchService) private readonly matches: MatchService) {} + @Post("bottles/pick") + pick( + @CurrentUser() user: AccessClaims, + @Headers("idempotency-key") key: string | undefined, + ) { + const normalized = key?.trim(); + if (!normalized || !/^[A-Za-z0-9][A-Za-z0-9._:-]{0,127}$/.test(normalized)) + throw new DomainException( + ErrorCode.VALIDATION_ERROR, + "Validation failed", + HttpStatus.BAD_REQUEST, + ); + return this.matches.pick(user.sub, normalized); + } + @Post("bottles/:id/return") + returnBottle( + @CurrentUser() user: AccessClaims, + @Param("id", new ParseUUIDPipe()) id: string, + @Body() dto: ReturnBottleDto, + ) { + return this.matches.returnBottle(user.sub, id, dto.leaseId, dto.token); + } +} +Reflect.defineMetadata( + "design:paramtypes", + [Object, String], + MatchController.prototype, + "pick", +); +Reflect.defineMetadata( + "design:paramtypes", + [Object, String, ReturnBottleDto], + MatchController.prototype, + "returnBottle", +); diff --git a/apps/api/src/match/match.e2e-spec.ts b/apps/api/src/match/match.e2e-spec.ts new file mode 100644 index 0000000..6bedd85 --- /dev/null +++ b/apps/api/src/match/match.e2e-spec.ts @@ -0,0 +1,434 @@ +/* eslint-disable @typescript-eslint/no-unsafe-member-access, @typescript-eslint/no-unsafe-assignment, @typescript-eslint/no-unsafe-argument, @typescript-eslint/no-unsafe-return */ +import "reflect-metadata"; +import { type INestApplication } from "@nestjs/common"; +import { Test } from "@nestjs/testing"; +import { PrismaClient } from "@prisma/client"; +import { randomUUID } from "node:crypto"; +import request from "supertest"; +import { afterAll, beforeAll, beforeEach, describe, expect, it } from "vitest"; +import { assertSafeTestDatabaseUrl } from "../../../../prisma/database-safety.js"; +import { AppModule } from "../app.module.js"; +import { TokenService } from "../auth/token.service.js"; +import { configureApp } from "../main.js"; + +process.env.NODE_ENV = "test"; +process.env.WEB_ORIGIN = "http://localhost:3000"; +process.env.PHONE_ENCRYPTION_KEY = + "AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA="; +process.env.PHONE_HMAC_KEY = "test-phone-hmac-key-with-at-least-32-bytes"; +process.env.VERIFICATION_CODE_HMAC_KEY = + "test-code-hmac-key-with-at-least-32-bytes"; +process.env.JWT_SECRET = "test-jwt-secret-with-at-least-thirty-two-bytes"; +process.env.REFRESH_TOKEN_HMAC_KEY = + "test-refresh-hmac-key-with-at-least-32-bytes"; +process.env.LEASE_TOKEN_HMAC_KEY = "test-lease-hmac-key-with-at-least-32-bytes"; +process.env.LEASE_TOKEN_ENCRYPTION_KEY = + "BBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBB="; + +const prisma = new PrismaClient(); + +describe("match API with real PostgreSQL", () => { + let app: INestApplication; + + beforeAll(async () => { + assertSafeTestDatabaseUrl(process.env.DATABASE_URL ?? ""); + await prisma.$connect(); + const module = await Test.createTestingModule({ + imports: [AppModule], + }).compile(); + app = module.createNestApplication(); + configureApp(app); + await app.init(); + }); + + beforeEach(async () => { + await prisma.$executeRawUnsafe(`TRUNCATE TABLE "accounts" CASCADE`); + }); + + afterAll(async () => { + await app?.close(); + await prisma.$disconnect(); + }); + + async function actor( + deviceId: string, + profileStatus: "APPROVED" | "REVIEWING" = "APPROVED", + ) { + const account = await prisma.account.create({ + data: { + phoneCiphertext: Buffer.from("cipher"), + phoneHmac: randomUUID(), + anonymousProfile: { + create: { + nickname: `anon-${deviceId}`, + avatarColor: "#123456", + bio: `bio-${deviceId}`, + reviewStatus: profileStatus, + }, + }, + }, + include: { anonymousProfile: true }, + }); + const session = await prisma.session.create({ + data: { + accountId: account.id, + refreshTokenHash: randomUUID(), + deviceId, + expiresAt: new Date(Date.now() + 60_000), + tokenFamily: randomUUID(), + }, + }); + const token = app.get(TokenService).issueAccess({ + sub: account.id, + session_id: session.id, + device_id: deviceId, + scopes: ["user"], + token_version: 0, + }); + return { + id: account.id, + authorization: `Bearer ${token}`, + profile: account.anonymousProfile!, + }; + } + + async function bottle(authorId: string, contentText = "message in bottle") { + return prisma.bottle.create({ + data: { + authorId, + clientRequestId: randomUUID(), + contentText, + reviewStatus: "APPROVED", + poolStatus: "IN_POOL", + approvedAt: new Date(), + }, + }); + } + + const pick = (authorization: string, key: string = randomUUID()) => + request(app.getHttpServer()) + .post("/api/v1/bottles/pick") + .set("Authorization", authorization) + .set("Idempotency-Key", key) + .send({}); + const returnBottle = ( + authorization: string, + bottleId: string, + leaseId: string, + token: string, + ) => + request(app.getHttpServer()) + .post(`/api/v1/bottles/${bottleId}/return`) + .set("Authorization", authorization) + .send({ leaseId, token }); + + it("picks an approved in-pool bottle and exposes only an anonymous public snapshot", async () => { + const author = await actor("author-device"); + const picker = await actor("picker-device"); + const source = await bottle(author.id, "hello from sea"); + const response = await pick(picker.authorization).expect(201); + expect(response.body.data).toMatchObject({ + bottle: { id: source.id, contentText: "hello from sea" }, + author: { + publicId: author.profile.publicId, + nickname: author.profile.nickname, + avatarColor: author.profile.avatarColor, + bio: author.profile.bio, + }, + lease: { + id: expect.any(String), + token: expect.stringMatching(/^[A-Za-z0-9_-]{32,}$/), + expiresAt: expect.any(String), + }, + }); + expect(JSON.stringify(response.body.data)).not.toContain(author.id); + expect(JSON.stringify(response.body.data)).not.toContain("phone"); + expect( + await prisma.bottle.findUniqueOrThrow({ where: { id: source.id } }), + ).toMatchObject({ poolStatus: "LEASED", version: 2 }); + expect( + await prisma.bottlePickHistory.count({ + where: { bottleId: source.id, pickerId: picker.id }, + }), + ).toBe(1); + expect( + ( + await prisma.dailyUsage.findFirstOrThrow({ + where: { accountId: picker.id }, + }) + ).bottlesPicked, + ).toBe(1); + }); + + it("excludes own, history, both block directions, unapproved profile and unavailable bottle states", async () => { + const picker = await actor("picker-device"); + const own = await bottle(picker.id, "own"); + const historyAuthor = await actor("history-author"); + const historical = await bottle(historyAuthor.id, "history"); + await prisma.bottlePickHistory.create({ + data: { bottleId: historical.id, pickerId: picker.id }, + }); + const blockedByPicker = await actor("blocked-one"); + await bottle(blockedByPicker.id, "blocked outgoing"); + await prisma.block.create({ + data: { blockerId: picker.id, blockedId: blockedByPicker.id }, + }); + const blocksPicker = await actor("blocked-two"); + await bottle(blocksPicker.id, "blocked incoming"); + await prisma.block.create({ + data: { blockerId: blocksPicker.id, blockedId: picker.id }, + }); + const pendingProfile = await actor("pending-profile", "REVIEWING"); + await bottle(pendingProfile.id, "pending profile"); + const validAuthor = await actor("valid-author"); + const valid = await bottle(validAuthor.id, "valid"); + await prisma.bottle.create({ + data: { + authorId: validAuthor.id, + clientRequestId: randomUUID(), + contentText: "reviewing", + reviewStatus: "REVIEWING", + poolStatus: "IN_POOL", + }, + }); + const picked = await pick(picker.authorization).expect(201); + expect(picked.body.data.bottle.id).toBe(valid.id); + expect(picked.body.data.bottle.id).not.toBe(own.id); + }); + + it("returns pool empty without consuming quota", async () => { + const picker = await actor("picker-device"); + const response = await pick(picker.authorization).expect(404); + expect(response.body.code).toBe("BOTTLE_POOL_EMPTY"); + expect( + await prisma.dailyUsage.count({ where: { accountId: picker.id } }), + ).toBe(0); + }); + + it("deduplicates concurrent pick retries into one lease and one quota use", async () => { + const author = await actor("author-device"); + const picker = await actor("picker-device"); + await bottle(author.id); + const key = randomUUID(); + const results = await Promise.all( + Array.from({ length: 8 }, () => pick(picker.authorization, key)), + ); + expect(results.every((x) => x.status === 201)).toBe(true); + expect(new Set(results.map((x) => x.body.data.lease.id))).toHaveLength(1); + expect(new Set(results.map((x) => x.body.data.lease.token))).toHaveLength( + 1, + ); + expect(await prisma.bottlePickLease.count()).toBe(1); + expect((await prisma.dailyUsage.findFirstOrThrow()).bottlesPicked).toBe(1); + const token = results[0]!.body.data.lease.token as string; + const stored = await prisma.bottlePickLease.findFirstOrThrow(); + expect(token).toMatch(/^[A-Za-z0-9_-]{43}$/); + expect(stored.leaseTokenHash).not.toContain(token); + expect( + Buffer.from(stored.leaseTokenCiphertext).toString("utf8"), + ).not.toContain(token); + }); + + it("fails closed with a stable response when an idempotent lease token cannot be decrypted", async () => { + const author = await actor("author-device"); + const picker = await actor("picker-device"); + await bottle(author.id); + const key = randomUUID(); + const first = await pick(picker.authorization, key).expect(201); + await prisma.bottlePickLease.update({ + where: { id: first.body.data.lease.id }, + data: { leaseTokenCiphertext: Buffer.from("corrupt") }, + }); + + const retry = await pick(picker.authorization, key).expect(503); + + expect(retry.body).toMatchObject({ + code: "SERVICE_UNAVAILABLE", + message: "Bottle pick temporarily unavailable", + }); + expect(JSON.stringify(retry.body)).not.toContain("cipher"); + expect(await prisma.bottlePickLease.count()).toBe(1); + expect((await prisma.dailyUsage.findFirstOrThrow()).bottlesPicked).toBe(1); + }); + + it("issues independent random opaque tokens for different requests", async () => { + const author = await actor("author-device"); + const picker = await actor("picker-device"); + await bottle(author.id, "one"); + await bottle(author.id, "two"); + const first = (await pick(picker.authorization, "request-one").expect(201)) + .body.data.lease.token; + const second = (await pick(picker.authorization, "request-two").expect(201)) + .body.data.lease.token; + expect(first).toMatch(/^[A-Za-z0-9_-]{43}$/); + expect(second).toMatch(/^[A-Za-z0-9_-]{43}$/); + expect(second).not.toBe(first); + }); + + it("rejects malformed return UUIDs and non-canonical lease tokens at the DTO boundary", async () => { + const picker = await actor("picker-device"); + await returnBottle( + picker.authorization, + "not-a-uuid", + randomUUID(), + "x".repeat(43), + ).expect(400); + await returnBottle( + picker.authorization, + randomUUID(), + "not-a-uuid", + "x".repeat(43), + ).expect(400); + await returnBottle( + picker.authorization, + randomUUID(), + randomUUID(), + "x".repeat(42), + ).expect(400); + }); + + it("rejects malformed idempotency keys and canonicalizes surrounding whitespace", async () => { + const author = await actor("author-device"); + const picker = await actor("picker-device"); + await bottle(author.id); + await pick(picker.authorization, "bad key").expect(400); + const first = await pick(picker.authorization, " stable-key ").expect(201); + const retry = await pick(picker.authorization, "stable-key").expect(201); + expect(retry.body.data.lease).toEqual(first.body.data.lease); + }); + + it("allows only one of two users to concurrently lease the sole bottle", async () => { + const author = await actor("author-device"); + const a = await actor("picker-a-device"); + const b = await actor("picker-b-device"); + const source = await bottle(author.id); + const results = await Promise.all([ + pick(a.authorization), + pick(b.authorization), + ]); + expect(results.filter((x) => x.status === 201)).toHaveLength(1); + expect( + results.filter( + (x) => x.status === 404 && x.body.code === "BOTTLE_POOL_EMPTY", + ), + ).toHaveLength(1); + expect( + await prisma.bottlePickLease.count({ + where: { bottleId: source.id, status: "ACTIVE" }, + }), + ).toBe(1); + }); + + it("enforces twenty successful picks atomically and resets on another UTC+8 day", async () => { + const author = await actor("author-device"); + const picker = await actor("picker-device"); + await Promise.all( + Array.from({ length: 22 }, (_, i) => bottle(author.id, `bottle-${i}`)), + ); + const results = await Promise.all( + Array.from({ length: 21 }, () => pick(picker.authorization)), + ); + expect(results.filter((x) => x.status === 201)).toHaveLength(20); + expect( + results.filter( + (x) => x.status === 429 && x.body.code === "BOTTLE_DAILY_LIMIT", + ), + ).toHaveLength(1); + expect((await prisma.dailyUsage.findFirstOrThrow()).bottlesPicked).toBe(20); + await prisma.dailyUsage.updateMany({ + data: { usageDate: new Date("2020-01-01T00:00:00.000Z") }, + }); + await pick(picker.authorization).expect(201); + }); + + it.each(["SUSPENSION", "BAN"] as const)( + "rejects active %s sanction", + async (type) => { + const author = await actor("author-device"); + const picker = await actor("picker-device"); + await bottle(author.id); + await prisma.sanction.create({ + data: { accountId: picker.id, type, reason: "policy" }, + }); + const response = await pick(picker.authorization).expect(403); + expect(response.body.code).toBe("ACCOUNT_SANCTIONED"); + expect(await prisma.bottlePickLease.count()).toBe(0); + }, + ); + + it("returns an active lease, is idempotent, and permanently preserves pick history", async () => { + const author = await actor("author-device"); + const picker = await actor("picker-device"); + const source = await bottle(author.id); + const picked = (await pick(picker.authorization).expect(201)).body.data; + const first = await returnBottle( + picker.authorization, + source.id, + picked.lease.id, + picked.lease.token, + ).expect(201); + const retry = await returnBottle( + picker.authorization, + source.id, + picked.lease.id, + picked.lease.token, + ).expect(201); + expect(first.body.data).toEqual({ + bottleId: source.id, + leaseId: picked.lease.id, + status: "RETURNED", + }); + expect(retry.body.data).toEqual(first.body.data); + expect( + await prisma.bottle.findUniqueOrThrow({ where: { id: source.id } }), + ).toMatchObject({ poolStatus: "IN_POOL", activeLeaseId: null, version: 3 }); + await pick(picker.authorization).expect(404); + }); + + it("hides another user's lease and rejects an invalid opaque token without mutation", async () => { + const author = await actor("author-device"); + const picker = await actor("picker-device"); + const stranger = await actor("stranger-device"); + const source = await bottle(author.id); + const picked = (await pick(picker.authorization).expect(201)).body.data; + await returnBottle( + stranger.authorization, + source.id, + picked.lease.id, + picked.lease.token, + ).expect(404); + await returnBottle( + picker.authorization, + source.id, + picked.lease.id, + "x".repeat(43), + ).expect(404); + expect( + await prisma.bottlePickLease.findUniqueOrThrow({ + where: { id: picked.lease.id }, + }), + ).toMatchObject({ status: "ACTIVE", endedAt: null }); + expect( + (await prisma.bottle.findUniqueOrThrow({ where: { id: source.id } })) + .poolStatus, + ).toBe("LEASED"); + }); + + it("rejects returning an expired lease without releasing it in the API", async () => { + const author = await actor("author-device"); + const picker = await actor("picker-device"); + const source = await bottle(author.id); + const picked = (await pick(picker.authorization).expect(201)).body.data; + await prisma.bottlePickLease.update({ + where: { id: picked.lease.id }, + data: { expiresAt: new Date(Date.now() - 1) }, + }); + const response = await returnBottle( + picker.authorization, + source.id, + picked.lease.id, + picked.lease.token, + ).expect(410); + expect(response.body.code).toBe("BOTTLE_LEASE_EXPIRED"); + }); +}); diff --git a/apps/api/src/match/match.module.ts b/apps/api/src/match/match.module.ts new file mode 100644 index 0000000..a36d979 --- /dev/null +++ b/apps/api/src/match/match.module.ts @@ -0,0 +1,13 @@ +import { Module } from "@nestjs/common"; +import { AuthModule } from "../auth/auth.module.js"; +import { DatabaseModule } from "../database/database.module.js"; +import { CandidateRepository } from "./candidate.repository.js"; +import { MatchController } from "./match.controller.js"; +import { MatchService } from "./match.service.js"; + +@Module({ + imports: [DatabaseModule, AuthModule], + controllers: [MatchController], + providers: [CandidateRepository, MatchService], +}) +export class MatchModule {} diff --git a/apps/api/src/match/match.service.ts b/apps/api/src/match/match.service.ts new file mode 100644 index 0000000..24e9be9 --- /dev/null +++ b/apps/api/src/match/match.service.ts @@ -0,0 +1,286 @@ +import { randomBytes, randomUUID } from "node:crypto"; +import { HttpStatus, Inject, Injectable } from "@nestjs/common"; +import { ErrorCode } from "@drift/contracts"; +import type { Prisma } from "@prisma/client"; +import { DomainException } from "../common/domain.exception.js"; +import { PrismaService } from "../database/prisma.service.js"; +import { utc8UsageDate } from "../bottle/usage-date.js"; +import { + decryptLeaseToken, + encryptLeaseToken, + leaseHmac, +} from "../auth/auth.config.js"; +import { CandidateRepository, type Candidate } from "./candidate.repository.js"; + +type LeaseWithBottle = { + id: string; + expiresAt: Date; + leaseTokenCiphertext: Uint8Array; + bottle: { + id: string; + contentText: string; + author: { + anonymousProfile: { + publicId: string; + nickname: string; + avatarColor: string; + bio: string | null; + } | null; + }; + }; +}; + +function positiveInt( + value: string | undefined, + fallback: number, + name: string, +) { + const parsed = value === undefined ? fallback : Number(value); + if (!Number.isInteger(parsed) || parsed <= 0) + throw new Error(`${name} must be a positive integer`); + return parsed; +} + +@Injectable() +export class MatchService { + private readonly ttlMs = positiveInt( + process.env.BOTTLE_LEASE_TTL_MS, + 300_000, + "BOTTLE_LEASE_TTL_MS", + ); + + constructor( + @Inject(PrismaService) private readonly prisma: PrismaService, + @Inject(CandidateRepository) + private readonly candidates: CandidateRepository, + ) {} + + async pick(pickerId: string, requestId: string) { + return this.prisma.$transaction( + async (tx) => { + await tx.$executeRaw`SELECT pg_advisory_xact_lock(hashtextextended(${`pick:${pickerId}`}, 0))`; + const previous = await tx.bottlePickRequest.findUnique({ + where: { pickerId_requestId: { pickerId, requestId } }, + include: { + lease: { + include: { + bottle: { + include: { author: { include: { anonymousProfile: true } } }, + }, + }, + }, + }, + }); + if (previous?.lease) { + let token: string; + try { + token = decryptLeaseToken( + previous.lease.leaseTokenCiphertext, + previous.lease.id, + ); + } catch { + throw new DomainException( + ErrorCode.SERVICE_UNAVAILABLE, + "Bottle pick temporarily unavailable", + HttpStatus.SERVICE_UNAVAILABLE, + ); + } + return this.response(previous.lease, token); + } + await this.assertEligible(tx, pickerId); + const usageDate = utc8UsageDate(new Date()); + const current = await tx.dailyUsage.findUnique({ + where: { + accountId_usageDate: { + accountId: pickerId, + usageDate: new Date(`${usageDate}T00:00:00.000Z`), + }, + }, + }); + if ((current?.bottlesPicked ?? 0) >= 20) this.limit(); + + const batch = await this.candidates.findBatch(tx, pickerId); + while (batch.length) { + const index = Math.floor(Math.random() * batch.length); + const candidate = batch.splice(index, 1)[0]!; + const result = await this.claim( + tx, + pickerId, + requestId, + usageDate, + candidate, + ); + if (result) return result; + } + throw new DomainException( + ErrorCode.BOTTLE_POOL_EMPTY, + "Bottle pool empty", + HttpStatus.NOT_FOUND, + ); + }, + { isolationLevel: "ReadCommitted" }, + ); + } + + private async claim( + tx: Prisma.TransactionClient, + pickerId: string, + requestId: string, + usageDate: string, + candidate: Candidate, + ) { + const token = randomBytes(32).toString("base64url"); + const expiresAt = new Date(Date.now() + this.ttlMs); + const leaseId = randomUUID(); + const claimed = await tx.bottle.updateMany({ + where: { + id: candidate.id, + version: candidate.version, + poolStatus: "IN_POOL", + activeLeaseId: null, + }, + data: { + poolStatus: "LEASED", + activeLeaseId: leaseId, + version: { increment: 1 }, + }, + }); + if (claimed.count !== 1) return null; + const lease = await tx.bottlePickLease.create({ + data: { + id: leaseId, + bottleId: candidate.id, + pickerId, + leaseTokenHash: leaseHmac(token), + leaseTokenCiphertext: encryptLeaseToken(token, leaseId), + expiresAt, + }, + }); + await tx.bottlePickHistory.create({ + data: { bottleId: candidate.id, pickerId }, + }); + const rows = await tx.$queryRaw>` + INSERT INTO "daily_usage" ("id", "account_id", "usage_date", "bottles_picked", "updated_at") + VALUES (gen_random_uuid(), ${pickerId}::uuid, ${usageDate}::date, 1, now()) + ON CONFLICT ("account_id", "usage_date") DO UPDATE + SET "bottles_picked" = "daily_usage"."bottles_picked" + 1, "updated_at" = now() + WHERE "daily_usage"."bottles_picked" < 20 + RETURNING "bottles_picked"`; + if (!rows.length) this.limit(); + await tx.bottlePickRequest.create({ + data: { pickerId, requestId, leaseId: lease.id }, + }); + return { + bottle: { id: candidate.id, contentText: candidate.contentText }, + author: { + publicId: candidate.publicId, + nickname: candidate.nickname, + avatarColor: candidate.avatarColor, + bio: candidate.bio, + }, + lease: { id: lease.id, token, expiresAt }, + }; + } + + private async assertEligible(tx: Prisma.TransactionClient, pickerId: string) { + const now = new Date(); + const account = await tx.account.findUnique({ + where: { id: pickerId }, + select: { status: true }, + }); + const sanction = await tx.sanction.findFirst({ + where: { + accountId: pickerId, + type: { in: ["SUSPENSION", "BAN"] }, + revokedAt: null, + startsAt: { lte: now }, + OR: [{ expiresAt: null }, { expiresAt: { gt: now } }], + }, + select: { id: true }, + }); + if (account?.status !== "ACTIVE" || sanction) + throw new DomainException( + ErrorCode.ACCOUNT_SANCTIONED, + "Account sanctioned", + HttpStatus.FORBIDDEN, + ); + } + + async returnBottle( + pickerId: string, + bottleId: string, + leaseId: string, + token: string, + ) { + const tokenHash = leaseHmac(token); + const hidden = () => + new DomainException( + ErrorCode.NOT_FOUND, + "Not Found", + HttpStatus.NOT_FOUND, + ); + return this.prisma.$transaction(async (tx) => { + const rows = await tx.$queryRaw< + Array<{ id: string; status: string; expiresAt: Date }> + >` + SELECT "id", "status", "expires_at" AS "expiresAt" + FROM "bottle_pick_leases" + WHERE "id" = ${leaseId}::uuid AND "bottle_id" = ${bottleId}::uuid + AND "picker_id" = ${pickerId}::uuid AND "lease_token_hash" = ${tokenHash} + FOR UPDATE`; + const lease = rows[0]; + if (!lease) throw hidden(); + if (lease.status === "RETURNED") + return { bottleId, leaseId, status: "RETURNED" as const }; + if (lease.status !== "ACTIVE") throw hidden(); + if (lease.expiresAt <= new Date()) + throw new DomainException( + ErrorCode.BOTTLE_LEASE_EXPIRED, + "Bottle lease expired", + HttpStatus.GONE, + ); + const released = await tx.bottle.updateMany({ + where: { id: bottleId, poolStatus: "LEASED", activeLeaseId: leaseId }, + data: { + poolStatus: "IN_POOL", + activeLeaseId: null, + version: { increment: 1 }, + }, + }); + if (released.count !== 1) throw hidden(); + await tx.bottlePickLease.update({ + where: { id: leaseId }, + data: { status: "RETURNED", endedAt: new Date() }, + }); + return { bottleId, leaseId, status: "RETURNED" as const }; + }); + } + + private response(lease: LeaseWithBottle, token: string) { + const profile = lease.bottle.author.anonymousProfile; + if (!profile) + throw new DomainException( + ErrorCode.BOTTLE_POOL_EMPTY, + "Bottle pool empty", + HttpStatus.NOT_FOUND, + ); + return { + bottle: { id: lease.bottle.id, contentText: lease.bottle.contentText }, + author: { + publicId: profile.publicId, + nickname: profile.nickname, + avatarColor: profile.avatarColor, + bio: profile.bio, + }, + lease: { id: lease.id, token, expiresAt: lease.expiresAt }, + }; + } + + private limit(): never { + throw new DomainException( + ErrorCode.BOTTLE_DAILY_LIMIT, + "Daily pick limit reached", + HttpStatus.TOO_MANY_REQUESTS, + ); + } +} diff --git a/apps/api/src/profile/profile.e2e-spec.ts b/apps/api/src/profile/profile.e2e-spec.ts index 4b06f10..d8be417 100644 --- a/apps/api/src/profile/profile.e2e-spec.ts +++ b/apps/api/src/profile/profile.e2e-spec.ts @@ -18,6 +18,9 @@ process.env.PHONE_ENCRYPTION_KEY = process.env.PHONE_HMAC_KEY = "test-phone-hmac-key-with-at-least-32-bytes"; process.env.VERIFICATION_CODE_HMAC_KEY = "test-code-hmac-key-with-at-least-32-bytes"; +process.env.LEASE_TOKEN_HMAC_KEY = "test-lease-hmac-key-with-at-least-32-bytes"; +process.env.LEASE_TOKEN_ENCRYPTION_KEY = + "BBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBB="; process.env.JWT_SECRET = "test-jwt-secret-with-at-least-thirty-two-bytes"; process.env.REFRESH_TOKEN_HMAC_KEY = "test-refresh-hmac-key-with-at-least-32-bytes"; diff --git a/apps/worker/src/lease-reaper.processor.ts b/apps/worker/src/lease-reaper.processor.ts new file mode 100644 index 0000000..b980b06 --- /dev/null +++ b/apps/worker/src/lease-reaper.processor.ts @@ -0,0 +1,40 @@ +import type { PrismaClient } from "@prisma/client"; + +export class LeaseReaper { + constructor(private readonly prisma: PrismaClient) {} + + async runOnce(): Promise { + return this.prisma.$transaction(async (tx) => { + const rows = await tx.$queryRaw>` + SELECT "id", "bottle_id" AS "bottleId" + FROM "bottle_pick_leases" + WHERE "status" = 'ACTIVE' AND "expires_at" <= now() + ORDER BY "expires_at", "id" + FOR UPDATE SKIP LOCKED LIMIT 1`; + const lease = rows[0]; + if (!lease) return false; + const expired = await tx.bottlePickLease.updateMany({ + where: { + id: lease.id, + status: "ACTIVE", + expiresAt: { lte: new Date() }, + }, + data: { status: "EXPIRED", endedAt: new Date() }, + }); + if (expired.count !== 1) return true; + await tx.bottle.updateMany({ + where: { + id: lease.bottleId, + poolStatus: "LEASED", + activeLeaseId: lease.id, + }, + data: { + poolStatus: "IN_POOL", + activeLeaseId: null, + version: { increment: 1 }, + }, + }); + return true; + }); + } +} diff --git a/apps/worker/src/lease-reaper.spec.ts b/apps/worker/src/lease-reaper.spec.ts new file mode 100644 index 0000000..778b345 --- /dev/null +++ b/apps/worker/src/lease-reaper.spec.ts @@ -0,0 +1,125 @@ +/* eslint-disable @typescript-eslint/no-unsafe-assignment */ +import { PrismaClient } from "@prisma/client"; +import { randomUUID } from "node:crypto"; +import { afterAll, beforeAll, beforeEach, describe, expect, it } from "vitest"; +import { assertSafeTestDatabaseUrl } from "../../../prisma/database-safety.js"; +import { LeaseReaper } from "./lease-reaper.processor.js"; + +const prisma = new PrismaClient(); + +describe("lease reaper with real PostgreSQL", () => { + beforeAll(async () => { + assertSafeTestDatabaseUrl(process.env.DATABASE_URL ?? ""); + await prisma.$connect(); + }); + beforeEach(async () => { + await prisma.$executeRawUnsafe(`TRUNCATE TABLE "accounts" CASCADE`); + }); + afterAll(() => prisma.$disconnect()); + + async function fixture(expiresAt: Date) { + const [author, picker] = await Promise.all([ + prisma.account.create({ + data: { phoneCiphertext: Buffer.from("a"), phoneHmac: randomUUID() }, + }), + prisma.account.create({ + data: { phoneCiphertext: Buffer.from("p"), phoneHmac: randomUUID() }, + }), + ]); + const bottle = await prisma.bottle.create({ + data: { + authorId: author.id, + clientRequestId: randomUUID(), + contentText: "secret body", + reviewStatus: "APPROVED", + poolStatus: "IN_POOL", + }, + }); + const leaseId = randomUUID(); + await prisma.$transaction(async (tx) => { + await tx.bottle.update({ + where: { id: bottle.id }, + data: { + poolStatus: "LEASED", + activeLeaseId: leaseId, + version: { increment: 1 }, + }, + }); + await tx.bottlePickLease.create({ + data: { + id: leaseId, + bottleId: bottle.id, + pickerId: picker.id, + leaseTokenHash: randomUUID(), + leaseTokenCiphertext: Buffer.from("ciphertext"), + expiresAt, + }, + }); + }); + return { bottle, leaseId }; + } + + it("atomically expires a due active lease and returns its current bottle to the pool", async () => { + const { bottle, leaseId } = await fixture(new Date(Date.now() - 1000)); + expect(await new LeaseReaper(prisma).runOnce()).toBe(true); + expect( + await prisma.bottlePickLease.findUniqueOrThrow({ + where: { id: leaseId }, + }), + ).toMatchObject({ status: "EXPIRED", endedAt: expect.any(Date) }); + expect( + await prisma.bottle.findUniqueOrThrow({ where: { id: bottle.id } }), + ).toMatchObject({ poolStatus: "IN_POOL", activeLeaseId: null, version: 3 }); + }); + + it("does not reap a future lease", async () => { + const { bottle, leaseId } = await fixture(new Date(Date.now() + 60_000)); + expect(await new LeaseReaper(prisma).runOnce()).toBe(false); + expect( + ( + await prisma.bottlePickLease.findUniqueOrThrow({ + where: { id: leaseId }, + }) + ).status, + ).toBe("ACTIVE"); + expect( + (await prisma.bottle.findUniqueOrThrow({ where: { id: bottle.id } })) + .activeLeaseId, + ).toBe(leaseId); + }); + + it("never releases a newer lease when processing stale old ownership", async () => { + const { bottle, leaseId: oldLeaseId } = await fixture( + new Date(Date.now() - 1000), + ); + const picker = await prisma.account.create({ + data: { phoneCiphertext: Buffer.from("n"), phoneHmac: randomUUID() }, + }); + const newLeaseId = randomUUID(); + await prisma.$transaction(async (tx) => { + await tx.bottlePickLease.update({ + where: { id: oldLeaseId }, + data: { status: "EXPIRED", endedAt: new Date() }, + }); + await tx.bottle.update({ + where: { id: bottle.id }, + data: { activeLeaseId: newLeaseId }, + }); + await tx.bottlePickLease.create({ + data: { + id: newLeaseId, + bottleId: bottle.id, + pickerId: picker.id, + leaseTokenHash: randomUUID(), + leaseTokenCiphertext: Buffer.from("ciphertext"), + expiresAt: new Date(Date.now() + 60_000), + }, + }); + }); + expect(await new LeaseReaper(prisma).runOnce()).toBe(false); + expect( + (await prisma.bottle.findUniqueOrThrow({ where: { id: bottle.id } })) + .activeLeaseId, + ).toBe(newLeaseId); + }); +}); diff --git a/apps/worker/src/main.spec.ts b/apps/worker/src/main.spec.ts index 7dc6595..be4bc94 100644 --- a/apps/worker/src/main.spec.ts +++ b/apps/worker/src/main.spec.ts @@ -1,7 +1,22 @@ import { describe, expect, it, vi } from "vitest"; -import { handleTopLevelError, logTopLevelError, runWorker } from "./main.js"; +import { + createCombinedWorker, + handleTopLevelError, + logTopLevelError, + runWorker, +} from "./main.js"; describe("worker lifecycle", () => { + it("runs both reaping and moderation every iteration without starvation", async () => { + const reaper = { runOnce: vi.fn().mockResolvedValue(true) }; + const moderation = { runOnce: vi.fn().mockResolvedValue(false) }; + const worker = createCombinedWorker(reaper, moderation); + + await expect(worker.runOnce()).resolves.toBe(true); + + expect(reaper.runOnce).toHaveBeenCalledTimes(1); + expect(moderation.runOnce).toHaveBeenCalledTimes(1); + }); it("waits for the current run before disconnecting on SIGTERM", async () => { let finish!: () => void; const runOnce = vi.fn( diff --git a/apps/worker/src/main.ts b/apps/worker/src/main.ts index 666c8f8..df895b1 100644 --- a/apps/worker/src/main.ts +++ b/apps/worker/src/main.ts @@ -1,6 +1,7 @@ import { pathToFileURL } from "node:url"; import { PrismaClient } from "@prisma/client"; import { ModerationWorker } from "./moderation-worker.js"; +import { LeaseReaper } from "./lease-reaper.processor.js"; type SignalSource = { once(signal: "SIGTERM" | "SIGINT", listener: () => void): unknown; @@ -16,6 +17,21 @@ type RunWorkerOptions = { signals?: SignalSource; }; +type Worker = { runOnce(): Promise }; + +export function createCombinedWorker( + reaper: Worker, + moderation: Worker, +): Worker { + return { + async runOnce() { + const reaped = await reaper.runOnce(); + const moderated = await moderation.runOnce(); + return reaped || moderated; + }, + }; +} + export async function runWorker(options: RunWorkerOptions) { const signals = options.signals ?? process; const sleep = @@ -47,7 +63,9 @@ export async function runWorker(options: RunWorkerOptions) { export async function main() { const prisma = new PrismaClient(); - const worker = new ModerationWorker(prisma); + const moderation = new ModerationWorker(prisma); + const reaper = new LeaseReaper(prisma); + const worker = createCombinedWorker(reaper, moderation); await runWorker({ worker, connect: () => prisma.$connect(), diff --git a/prisma/migrations/0008_bottle_pick_leases/migration.sql b/prisma/migrations/0008_bottle_pick_leases/migration.sql new file mode 100644 index 0000000..00794ff --- /dev/null +++ b/prisma/migrations/0008_bottle_pick_leases/migration.sql @@ -0,0 +1,36 @@ +ALTER TABLE "bottles" ADD COLUMN "active_lease_id" UUID; +-- Legacy leases predate recoverable idempotency tokens. Expire them rather than +-- fabricating ciphertext that could ever be returned as a valid bearer secret. +ALTER TABLE "bottle_pick_leases" ADD COLUMN "lease_token_ciphertext" BYTEA; +WITH expired AS ( + UPDATE "bottle_pick_leases" + SET "status" = 'EXPIRED', "ended_at" = COALESCE("ended_at", CURRENT_TIMESTAMP) + WHERE "status" = 'ACTIVE' + RETURNING "bottle_id" +) +UPDATE "bottles" b +SET "pool_status" = 'IN_POOL', "version" = b."version" + 1 +WHERE b."pool_status" = 'LEASED' + AND b."id" IN (SELECT "bottle_id" FROM expired); +UPDATE "bottle_pick_leases" +SET "lease_token_ciphertext" = decode(repeat('00', 48), 'hex') +WHERE "lease_token_ciphertext" IS NULL; +ALTER TABLE "bottle_pick_leases" + ALTER COLUMN "lease_token_ciphertext" SET NOT NULL; + +CREATE TABLE "bottle_pick_requests" ( + "id" UUID NOT NULL DEFAULT gen_random_uuid(), + "picker_id" UUID NOT NULL, + "request_id" VARCHAR(128) NOT NULL, + "lease_id" UUID, + "created_at" TIMESTAMPTZ(3) NOT NULL DEFAULT CURRENT_TIMESTAMP, + CONSTRAINT "bottle_pick_requests_pkey" PRIMARY KEY ("id") +); + +CREATE UNIQUE INDEX "bottles_active_lease_id_key" ON "bottles"("active_lease_id"); +CREATE UNIQUE INDEX "bottle_pick_requests_picker_id_request_id_key" ON "bottle_pick_requests"("picker_id", "request_id"); +CREATE UNIQUE INDEX "bottle_pick_requests_lease_id_key" ON "bottle_pick_requests"("lease_id"); +CREATE INDEX "bottle_pick_requests_created_at_idx" ON "bottle_pick_requests"("created_at"); +ALTER TABLE "bottles" ADD CONSTRAINT "bottles_active_lease_id_fkey" FOREIGN KEY ("active_lease_id") REFERENCES "bottle_pick_leases"("id") ON DELETE SET NULL ON UPDATE CASCADE DEFERRABLE INITIALLY DEFERRED; +ALTER TABLE "bottle_pick_requests" ADD CONSTRAINT "bottle_pick_requests_picker_id_fkey" FOREIGN KEY ("picker_id") REFERENCES "accounts"("id") ON DELETE CASCADE ON UPDATE CASCADE; +ALTER TABLE "bottle_pick_requests" ADD CONSTRAINT "bottle_pick_requests_lease_id_fkey" FOREIGN KEY ("lease_id") REFERENCES "bottle_pick_leases"("id") ON DELETE SET NULL ON UPDATE CASCADE; \ No newline at end of file diff --git a/prisma/schema.prisma b/prisma/schema.prisma index 170797c..2cf4699 100644 --- a/prisma/schema.prisma +++ b/prisma/schema.prisma @@ -118,6 +118,7 @@ model Account { notifications Notification[] auditLogs AuditLog[] @relation("AuditActor") dailyUsage DailyUsage[] + pickRequests BottlePickRequest[] @@map("accounts") } @@ -213,6 +214,8 @@ model Bottle { pickHistory BottlePickHistory[] conversation Conversation? reports Report[] + activeLeaseId String? @unique @map("active_lease_id") @db.Uuid + activeLease BottlePickLease? @relation("ActiveBottleLease", fields: [activeLeaseId], references: [id], onDelete: SetNull) @@unique([authorId, clientRequestId]) @@index([poolStatus, createdAt]) @@ -223,22 +226,39 @@ model Bottle { model BottlePickLease { /// Partial unique index bottle_pick_leases_one_active_per_bottle is managed in 0001_init SQL. - id String @id @default(uuid()) @db.Uuid - bottleId String @map("bottle_id") @db.Uuid - pickerId String @map("picker_id") @db.Uuid - leaseTokenHash String @unique @map("lease_token_hash") @db.VarChar(255) - expiresAt DateTime @map("expires_at") @db.Timestamptz(3) - status BottlePickLeaseStatus @default(ACTIVE) - endedAt DateTime? @map("ended_at") @db.Timestamptz(3) - createdAt DateTime @default(now()) @map("created_at") @db.Timestamptz(3) - bottle Bottle @relation(fields: [bottleId], references: [id], onDelete: Cascade) - picker Account @relation("PickerLeases", fields: [pickerId], references: [id], onDelete: Cascade) + id String @id @default(uuid()) @db.Uuid + bottleId String @map("bottle_id") @db.Uuid + pickerId String @map("picker_id") @db.Uuid + leaseTokenHash String @unique @map("lease_token_hash") @db.VarChar(255) + leaseTokenCiphertext Bytes @map("lease_token_ciphertext") + expiresAt DateTime @map("expires_at") @db.Timestamptz(3) + status BottlePickLeaseStatus @default(ACTIVE) + endedAt DateTime? @map("ended_at") @db.Timestamptz(3) + createdAt DateTime @default(now()) @map("created_at") @db.Timestamptz(3) + bottle Bottle @relation(fields: [bottleId], references: [id], onDelete: Cascade) + activeForBottle Bottle? @relation("ActiveBottleLease") + picker Account @relation("PickerLeases", fields: [pickerId], references: [id], onDelete: Cascade) + pickRequest BottlePickRequest? @@index([pickerId, expiresAt]) @@index([bottleId, expiresAt]) @@map("bottle_pick_leases") } +model BottlePickRequest { + id String @id @default(uuid()) @db.Uuid + pickerId String @map("picker_id") @db.Uuid + requestId String @map("request_id") @db.VarChar(128) + leaseId String? @unique @map("lease_id") @db.Uuid + createdAt DateTime @default(now()) @map("created_at") @db.Timestamptz(3) + picker Account @relation(fields: [pickerId], references: [id], onDelete: Cascade) + lease BottlePickLease? @relation(fields: [leaseId], references: [id], onDelete: SetNull) + + @@unique([pickerId, requestId]) + @@index([createdAt]) + @@map("bottle_pick_requests") +} + model BottlePickHistory { id String @id @default(uuid()) @db.Uuid bottleId String @map("bottle_id") @db.Uuid diff --git a/tests/integration/database.spec.ts b/tests/integration/database.spec.ts index b51ca31..bcbe5be 100644 --- a/tests/integration/database.spec.ts +++ b/tests/integration/database.spec.ts @@ -165,6 +165,7 @@ describe("database authority constraints", () => { bottleId: bottle.id, pickerId: firstPicker.id, leaseTokenHash: "lease-a", + leaseTokenCiphertext: Buffer.from("ciphertext-a"), expiresAt: new Date(Date.now() + 60_000), }, }); @@ -174,6 +175,7 @@ describe("database authority constraints", () => { bottleId: bottle.id, pickerId: secondPicker.id, leaseTokenHash: "lease-b", + leaseTokenCiphertext: Buffer.from("ciphertext-b"), expiresAt: new Date(Date.now() + 60_000), }, }), @@ -188,6 +190,7 @@ describe("database authority constraints", () => { bottleId: bottle.id, pickerId: secondPicker.id, leaseTokenHash: "lease-c", + leaseTokenCiphertext: Buffer.from("ciphertext-c"), expiresAt: new Date(Date.now() + 60_000), }, }),