mirror of
https://github.com/jjbrunton/Sauci.git
synced 2026-10-31 18:19:43 +00:00
perf(app): reduce redundant data fetching
Share bounded database pools across API repositories, consolidate sync and chat queries, and make mobile refreshes cache-aware and foreground-aware. Add focused regression coverage, client freshness documentation, and repository agent role configuration.
This commit is contained in:
1 parent
9957fa43f2
commit
c7ee8a4851
81 files changed
+4456
-1400
No files matched your search
@@ -3,6 +3,14 @@ DATABASE_URL=postgresql://sauci:sauci_local_only@127.0.0.1:54320/sauci
|
||||
SUPABASE_AUTH_URL=https://your-project.supabase.co
|
||||
SUPABASE_AUTH_AUDIENCE=authenticated
|
||||
|
||||
# Optional connection-pool bounds. The API process opens one pool and shares it
|
||||
# across every repository, so these are the process-wide limits, not per-domain
|
||||
# ones. Raising DATABASE_POOL_MAX raises the server connections this instance can
|
||||
# hold; keep the total across instances inside the database's own limit.
|
||||
# DATABASE_POOL_MAX=10
|
||||
# DATABASE_POOL_IDLE_TIMEOUT_MS=30000
|
||||
# DATABASE_POOL_CONNECTION_TIMEOUT_MS=10000
|
||||
|
||||
# Server-only hosted Auth Admin credential. Used for account deletion and for
|
||||
# enriching admin user lists; it is never exposed to the browser or used as a
|
||||
# Supabase database client.
|
||||
|
||||
@@ -7,6 +7,9 @@ const baseSchema = z.object({
|
||||
NODE_ENV: z.enum(['development', 'test', 'production']).default('development'),
|
||||
PORT: z.coerce.number().int().min(1).max(65535).default(3003),
|
||||
DATABASE_URL: z.string().url(),
|
||||
DATABASE_POOL_MAX: z.coerce.number().int().min(1).max(200).default(10),
|
||||
DATABASE_POOL_IDLE_TIMEOUT_MS: z.coerce.number().int().min(0).max(3_600_000).default(30_000),
|
||||
DATABASE_POOL_CONNECTION_TIMEOUT_MS: z.coerce.number().int().min(0).max(60_000).default(10_000),
|
||||
SUPABASE_AUTH_URL: z.string().url().optional(),
|
||||
SUPABASE_AUTH_ISSUER: z.string().url().optional(),
|
||||
SUPABASE_AUTH_JWKS_URL: z.string().url().optional(),
|
||||
@@ -28,6 +31,7 @@ export interface AppConfig {
|
||||
nodeEnv: 'development' | 'test' | 'production';
|
||||
port: number;
|
||||
databaseUrl: string;
|
||||
databasePool: { max: number; idleTimeoutMillis: number; connectionTimeoutMillis: number };
|
||||
authIssuer: string;
|
||||
authJwksUrl?: string;
|
||||
authAudience: string;
|
||||
@@ -81,6 +85,11 @@ export function loadConfig(env: NodeJS.ProcessEnv = process.env): AppConfig {
|
||||
nodeEnv: parsed.NODE_ENV,
|
||||
port: parsed.PORT,
|
||||
databaseUrl: parsed.DATABASE_URL,
|
||||
databasePool: {
|
||||
max: parsed.DATABASE_POOL_MAX,
|
||||
idleTimeoutMillis: parsed.DATABASE_POOL_IDLE_TIMEOUT_MS,
|
||||
connectionTimeoutMillis: parsed.DATABASE_POOL_CONNECTION_TIMEOUT_MS,
|
||||
},
|
||||
authIssuer,
|
||||
authJwksUrl,
|
||||
authAudience: parsed.SUPABASE_AUTH_AUDIENCE,
|
||||
|
||||
@@ -0,0 +1,50 @@
|
||||
import { Pool } from 'pg';
|
||||
|
||||
/**
|
||||
* A repository is constructed either from a connection string, in which case it
|
||||
* owns the pool it creates and must end it, or from a pool the caller owns. The
|
||||
* API process takes the second path so every domain shares one bounded pool
|
||||
* instead of opening a default-sized pool per repository; tests keep taking the
|
||||
* first so each suite stays independently closable.
|
||||
*/
|
||||
export type DatabaseConnection = string | Pool;
|
||||
|
||||
export interface DatabasePoolOptions {
|
||||
/** Server-wide ceiling on concurrent PostgreSQL connections. */
|
||||
max?: number;
|
||||
idleTimeoutMillis?: number;
|
||||
connectionTimeoutMillis?: number;
|
||||
}
|
||||
|
||||
/**
|
||||
* `pg` defaults to ten connections per pool, which multiplied by one pool per
|
||||
* repository. One shared pool of the same size keeps the same headroom for a
|
||||
* single API process while bounding what a self-hosted PostgreSQL must accept.
|
||||
*/
|
||||
export const DEFAULT_POOL_MAX = 10;
|
||||
|
||||
export function createDatabasePool(databaseUrl: string, options: DatabasePoolOptions = {}): Pool {
|
||||
return new Pool({
|
||||
connectionString: databaseUrl,
|
||||
max: options.max ?? DEFAULT_POOL_MAX,
|
||||
...(options.idleTimeoutMillis === undefined ? {} : { idleTimeoutMillis: options.idleTimeoutMillis }),
|
||||
...(options.connectionTimeoutMillis === undefined ? {} : { connectionTimeoutMillis: options.connectionTimeoutMillis }),
|
||||
});
|
||||
}
|
||||
|
||||
export interface ResolvedPool {
|
||||
pool: Pool;
|
||||
/** True when this repository created the pool and is responsible for ending it. */
|
||||
owned: boolean;
|
||||
}
|
||||
|
||||
export function resolvePool(connection: DatabaseConnection): ResolvedPool {
|
||||
return typeof connection === 'string'
|
||||
? { pool: new Pool({ connectionString: connection }), owned: true }
|
||||
: { pool: connection, owned: false };
|
||||
}
|
||||
|
||||
/** Ends the pool only when this repository created it, so shared pools survive. */
|
||||
export async function closeResolvedPool(pool: Pool, owned: boolean): Promise<void> {
|
||||
if (owned) await pool.end();
|
||||
}
|
||||
@@ -2,6 +2,7 @@ import type { FeatureInterestResponse, Profile } from '@sauci/shared';
|
||||
import { and, eq, sql } from 'drizzle-orm';
|
||||
import { drizzle, type NodePgDatabase } from 'drizzle-orm/node-postgres';
|
||||
import { Pool } from 'pg';
|
||||
import { closeResolvedPool, resolvePool, type DatabaseConnection } from './pool.js';
|
||||
import type { AuthIdentity } from '../auth.js';
|
||||
import { featureInterests, profiles, type ProfileRow } from './schema.js';
|
||||
|
||||
@@ -44,10 +45,13 @@ function toProfile(row: ProfileRow): MobileCompatibleProfile {
|
||||
|
||||
export class PostgresRepository implements ApiRepository {
|
||||
private readonly pool: Pool;
|
||||
private readonly ownsPool: boolean;
|
||||
private readonly db: NodePgDatabase;
|
||||
|
||||
constructor(databaseUrl: string) {
|
||||
this.pool = new Pool({ connectionString: databaseUrl });
|
||||
constructor(connection: DatabaseConnection) {
|
||||
const resolved = resolvePool(connection);
|
||||
this.pool = resolved.pool;
|
||||
this.ownsPool = resolved.owned;
|
||||
this.db = drizzle(this.pool);
|
||||
}
|
||||
|
||||
@@ -55,7 +59,21 @@ export class PostgresRepository implements ApiRepository {
|
||||
await this.db.select({ id: profiles.id }).from(profiles).limit(1);
|
||||
}
|
||||
|
||||
/**
|
||||
* Idempotent for the common case. `GET /v1/me` runs on every app start and on
|
||||
* every session refresh, and the previous unconditional upsert made each of
|
||||
* those a write that advanced `auth_synced_at` and produced dead row versions
|
||||
* for autovacuum. The only fields this sync can actually change are the ones
|
||||
* the conflict clause sets, so a profile that already carries the identity's
|
||||
* email is returned as read and nothing is written. First-use creation and
|
||||
* email enrichment keep their existing behaviour.
|
||||
*/
|
||||
async upsertProfile(identity: AuthIdentity): Promise<MobileCompatibleProfile> {
|
||||
const [existing] = await this.db.select().from(profiles).where(eq(profiles.id, identity.id)).limit(1);
|
||||
if (existing && (identity.email === null || identity.email === undefined || existing.email === identity.email)) {
|
||||
return toProfile(existing);
|
||||
}
|
||||
|
||||
const [row] = await this.db.insert(profiles).values({
|
||||
id: identity.id,
|
||||
name: identity.name,
|
||||
@@ -93,6 +111,6 @@ export class PostgresRepository implements ApiRepository {
|
||||
}
|
||||
|
||||
async close(): Promise<void> {
|
||||
await this.pool.end();
|
||||
await closeResolvedPool(this.pool, this.ownsPool);
|
||||
}
|
||||
}
|
||||
@@ -1,4 +1,5 @@
|
||||
import { Pool, type PoolClient, type QueryResultRow } from 'pg';
|
||||
import { closeResolvedPool, resolvePool, type DatabaseConnection } from '../../db/pool.js';
|
||||
import { AccountOperationError } from './types.js';
|
||||
|
||||
interface ProfileOperationRecord extends QueryResultRow {
|
||||
@@ -63,9 +64,12 @@ async function lockCoupleAndPartner(
|
||||
|
||||
export class PostgresAccountOperationsRepository implements AccountOperationsRepository {
|
||||
private readonly pool: Pool;
|
||||
private readonly ownsPool: boolean;
|
||||
|
||||
constructor(databaseUrl: string) {
|
||||
this.pool = new Pool({ connectionString: databaseUrl });
|
||||
constructor(connection: DatabaseConnection) {
|
||||
const resolved = resolvePool(connection);
|
||||
this.pool = resolved.pool;
|
||||
this.ownsPool = resolved.owned;
|
||||
}
|
||||
|
||||
async deleteRelationship(userId: string): Promise<DestructiveOperationResult> {
|
||||
@@ -135,7 +139,7 @@ export class PostgresAccountOperationsRepository implements AccountOperationsRep
|
||||
}
|
||||
|
||||
async close(): Promise<void> {
|
||||
await this.pool.end();
|
||||
await closeResolvedPool(this.pool, this.ownsPool);
|
||||
}
|
||||
}
|
||||
|
||||
@@ -1,4 +1,5 @@
|
||||
import { Pool, type PoolClient, type QueryResultRow } from 'pg';
|
||||
import { closeResolvedPool, resolvePool, type DatabaseConnection } from '../../db/pool.js';
|
||||
import { readFile } from 'node:fs/promises';
|
||||
import { resolve, sep } from 'node:path';
|
||||
import { AdminError, type AdminPrincipal, type AdminQuery } from './types.js';
|
||||
@@ -139,9 +140,10 @@ export interface AdminRepository {
|
||||
|
||||
export class PostgresAdminRepository implements AdminRepository {
|
||||
private readonly pool: Pool;
|
||||
constructor(databaseUrl: string, private readonly options: {
|
||||
private readonly ownsPool: boolean;
|
||||
constructor(connection: DatabaseConnection, private readonly options: {
|
||||
adminPrivateKeyJwk?: string; mediaRoot?: string; authDirectory?: AdminAuthDirectory;
|
||||
} = {}) { this.pool = new Pool({ connectionString: databaseUrl }); }
|
||||
} = {}) { const resolved = resolvePool(connection); this.pool = resolved.pool; this.ownsPool = resolved.owned; }
|
||||
|
||||
async principal(userId: string): Promise<AdminPrincipal> {
|
||||
const result = await this.pool.query<{ id: string; user_id: string; role: AdminPrincipal['role']; permissions: string[] }>(
|
||||
@@ -356,5 +358,5 @@ export class PostgresAdminRepository implements AdminRepository {
|
||||
await client.query('insert into audit_logs(admin_user_id,actor_user_id,table_name,action,record_id,old_values,new_values) values($1,$2,$3,$4,$5,$6,$7)', [principal.adminId, principal.userId, table, action, recordId || null, redactAudit(oldValues), redactAudit(newValues)]);
|
||||
}
|
||||
|
||||
async close(): Promise<void> { await this.pool.end(); }
|
||||
async close(): Promise<void> { await closeResolvedPool(this.pool, this.ownsPool); }
|
||||
}
|
||||
@@ -1,4 +1,5 @@
|
||||
import { Pool, type PoolClient } from 'pg';
|
||||
import { closeResolvedPool, resolvePool, type DatabaseConnection } from '../../db/pool.js';
|
||||
import { AnswersError, calculateMatchType, type Answer, type MatchType, type QuestionType, type ResponseData } from './types.js';
|
||||
|
||||
interface ContextRow { couple_id: string | null; gender: string | null; max_intensity: number; is_premium: boolean }
|
||||
@@ -44,7 +45,7 @@ export interface AnswersRepository {
|
||||
submit(userId: string, input: SubmitInput): Promise<{ response: ResponseRow; match: (MatchRow & { question: QuestionRow }) | null }>;
|
||||
update(userId: string, input: UpdateInput): Promise<Record<string, unknown>>;
|
||||
responses(userId: string, page: number, limit: number): Promise<{ responses: unknown[]; totalCount: number }>;
|
||||
matches(userId: string, page: number, limit: number, archived: boolean): Promise<{ matches: unknown[]; totalCount: number }>;
|
||||
matches(userId: string, page: number, limit: number, archived: boolean): Promise<{ matches: unknown[]; totalCount: number | null; hasMore: boolean }>;
|
||||
markSeen(userId: string, ids: string[]): Promise<void>;
|
||||
archive(userId: string, matchId: string, archived: boolean): Promise<void>;
|
||||
streak(userId: string): Promise<CoupleStreakView | null>;
|
||||
@@ -53,8 +54,9 @@ export interface AnswersRepository {
|
||||
|
||||
export class PostgresAnswersRepository implements AnswersRepository {
|
||||
private readonly pool: Pool;
|
||||
private readonly ownsPool: boolean;
|
||||
private zones: Promise<Set<string>> | null = null;
|
||||
constructor(databaseUrl: string) { this.pool = new Pool({ connectionString: databaseUrl }); }
|
||||
constructor(connection: DatabaseConnection) { const resolved = resolvePool(connection); this.pool = resolved.pool; this.ownsPool = resolved.owned; }
|
||||
|
||||
private async context(client: Pool | PoolClient, userId: string): Promise<ContextRow> {
|
||||
const result = await client.query<ContextRow>('select couple_id, gender, max_intensity, is_premium from profiles where id=$1', [userId]);
|
||||
@@ -230,12 +232,23 @@ export class PostgresAnswersRepository implements AnswersRepository {
|
||||
return {responses:rows.rows.map(this.dates),totalCount:count.rows[0]?.count??0};
|
||||
}
|
||||
|
||||
/**
|
||||
* Reading one row past the page decides `hasMore` exactly, and the total — which
|
||||
* the client only records when it refreshes — is counted on the first page only
|
||||
* instead of on every scroll. Unread is aggregated once per couple rather than as
|
||||
* a correlated count per row: the sort key depends on it, so the old subquery
|
||||
* already ran for every candidate match, not just the twenty that were returned.
|
||||
*/
|
||||
async matches(userId:string,page:number,limit:number,archived:boolean){
|
||||
const ctx=await this.context(this.pool,userId); const offset=page*limit;
|
||||
const params=[userId,ctx.couple_id,archived,limit,offset];
|
||||
const base=`from matches m join questions q on q.id=m.question_id left join match_archives a on a.match_id=m.id and a.user_id=$1 where m.couple_id=$2 and (($3 and a.id is not null) or (not $3 and a.id is null))`;
|
||||
const [rows,count]=await Promise.all([this.pool.query(`select m.*,json_build_object('id',q.id,'pack_id',q.pack_id,'text',q.text,'partner_text',q.partner_text,'intensity',q.intensity,'question_type',q.question_type,'config',q.config,'created_at',q.created_at) question,(select count(*)::int from messages msg where msg.match_id=m.id and msg.user_id<>$1 and msg.read_at is null) "unreadCount" ${base} order by "unreadCount" desc,m.created_at desc limit $4 offset $5`,params),this.pool.query<{count:number}>(`select count(*)::int count ${base}`,params.slice(0,3))]);
|
||||
return {matches:rows.rows.map(this.dates),totalCount:count.rows[0]?.count??0};
|
||||
const params=[userId,ctx.couple_id,archived,limit+1,offset];
|
||||
const from=`from matches m join questions q on q.id=m.question_id left join match_archives a on a.match_id=m.id and a.user_id=$1`;
|
||||
const where=`where m.couple_id=$2 and (($3 and a.id is not null) or (not $3 and a.id is null))`;
|
||||
const unread=`unread as (select msg.match_id,count(*)::int c from messages msg join matches um on um.id=msg.match_id and um.couple_id=$2 where msg.user_id<>$1 and msg.read_at is null group by msg.match_id)`;
|
||||
const [rows,count]=await Promise.all([
|
||||
this.pool.query(`with ${unread} select m.*,json_build_object('id',q.id,'pack_id',q.pack_id,'text',q.text,'partner_text',q.partner_text,'intensity',q.intensity,'question_type',q.question_type,'config',q.config,'created_at',q.created_at) question,coalesce(u.c,0) "unreadCount" ${from} left join unread u on u.match_id=m.id ${where} order by "unreadCount" desc,m.created_at desc limit $4 offset $5`,params),
|
||||
page===0?this.pool.query<{count:number}>(`select count(*)::int count ${from} ${where}`,params.slice(0,3)):null]);
|
||||
return {matches:rows.rows.slice(0,limit).map(this.dates),totalCount:count?count.rows[0]?.count??0:null,hasMore:rows.rows.length>limit};
|
||||
}
|
||||
async markSeen(userId:string,ids:string[]){const ctx=await this.context(this.pool,userId);await this.pool.query('update matches set is_new=false where couple_id=$1 and id=any($2::uuid[])',[ctx.couple_id,ids]);}
|
||||
async archive(userId:string,matchId:string,archived:boolean){const ctx=await this.context(this.pool,userId);const found=await this.pool.query('select 1 from matches where id=$1 and couple_id=$2',[matchId,ctx.couple_id]);if(!found.rowCount)throw new AnswersError('match_not_found',404,'Match was not found');if(archived)await this.pool.query('insert into match_archives(match_id,user_id) values($1,$2) on conflict do nothing',[matchId,userId]);else await this.pool.query('delete from match_archives where match_id=$1 and user_id=$2',[matchId,userId]);}
|
||||
@@ -304,5 +317,5 @@ export class PostgresAnswersRepository implements AnswersRepository {
|
||||
}
|
||||
private async tx<T>(fn:(client:PoolClient)=>Promise<T>):Promise<T>{const c=await this.pool.connect();try{await c.query('begin');const value=await fn(c);await c.query('commit');return value;}catch(e){await c.query('rollback');throw e;}finally{c.release();}}
|
||||
private dates<T extends Record<string,any>>(row:T):T{return Object.fromEntries(Object.entries(row).map(([k,v])=>[k,v instanceof Date?v.toISOString():v])) as T;}
|
||||
async close(){await this.pool.end();}
|
||||
async close(){await closeResolvedPool(this.pool,this.ownsPool);}
|
||||
}
|
||||
@@ -1,4 +1,5 @@
|
||||
import { Pool, type QueryResultRow } from 'pg';
|
||||
import { closeResolvedPool, resolvePool, type DatabaseConnection } from '../../db/pool.js';
|
||||
|
||||
export interface PackContext { id: string; name: string; icon: string | null }
|
||||
export interface AppQuestion {
|
||||
@@ -19,6 +20,50 @@ export interface StrokeSegment {
|
||||
}
|
||||
export interface LiveDrawState { strokes: StrokeSegment[]; revision: number; updated_at: string | null; updated_by: string | null }
|
||||
|
||||
/**
|
||||
* A change summary rather than a payload. The mobile app used to re-fetch profile,
|
||||
* couple, matches, both pending directions and enabled packs every five seconds to
|
||||
* notice partner activity; it now compares this summary with the last one it saw and
|
||||
* refreshes only the domains whose marker actually moved. Every field is a marker,
|
||||
* not display data, so the response stays a few hundred bytes whatever the couple's
|
||||
* history looks like.
|
||||
*/
|
||||
export interface SyncSummary {
|
||||
server_time: string;
|
||||
couple_id: string | null;
|
||||
profile_updated_at: string | null;
|
||||
partner_id: string | null;
|
||||
partner_updated_at: string | null;
|
||||
/** Matches visible to this member: archiving one of them moves the count. */
|
||||
match_count: number;
|
||||
new_match_count: number;
|
||||
latest_match_at: string | null;
|
||||
/** Partner answered, this member has not. */
|
||||
pending_yours: number;
|
||||
/** This member answered, partner has not. */
|
||||
pending_theirs: number;
|
||||
unread_total: number;
|
||||
/** Digest of the enabled pack ids, so an equal-and-opposite toggle is still a change. */
|
||||
enabled_packs_fingerprint: string;
|
||||
/**
|
||||
* Digest of each visible match's type and response summary. `match_count` /
|
||||
* `new_match_count` / `latest_match_at` only move when a match is created,
|
||||
* archived or unarchived — not when a partner edits an existing response and
|
||||
* changes its `match_type` or `response_summary` — so this fingerprint is
|
||||
* what actually notices that edit.
|
||||
*/
|
||||
match_state_fingerprint: string;
|
||||
/**
|
||||
* Digest of each visible match's unread count. `unread_total` can stay equal
|
||||
* while unread messages redistribute between matches (one read, another
|
||||
* receives a message), which reorders the unread-first list without moving
|
||||
* the total; this fingerprint is what notices that.
|
||||
*/
|
||||
match_unread_fingerprint: string;
|
||||
/** couple_streaks.updated_at, so a partner's answer can silently refresh an already-loaded streak. */
|
||||
streak_updated_at: string | null;
|
||||
}
|
||||
|
||||
export class AppDataError extends Error {
|
||||
constructor(public readonly code: 'not_found' | 'no_couple' | 'not_recipient' | 'rate_limited' | 'revision_conflict', message: string, public readonly status: 403 | 404 | 409 | 429, public readonly details?: Record<string, unknown>) { super(message); }
|
||||
}
|
||||
@@ -31,6 +76,7 @@ export interface AppDataRepository {
|
||||
markMediaViewed(userId: string, messageId: string, expiresAt: string | null): Promise<{ media_viewed_at: string; media_expires_at: string | null }>;
|
||||
nudgeStatus(userId: string): Promise<{ last_nudge_sent_at: string | null }>;
|
||||
sendNudge(userId: string): Promise<{ success: true; notification_sent: boolean; reason?: string; next_nudge_available_at: string }>;
|
||||
syncSummary(userId: string): Promise<SyncSummary>;
|
||||
getLiveDraw(userId: string): Promise<LiveDrawState>;
|
||||
putLiveDraw(userId: string, strokes: StrokeSegment[], baseRevision: number): Promise<LiveDrawState>;
|
||||
close(): Promise<void>;
|
||||
@@ -38,11 +84,22 @@ export interface AppDataRepository {
|
||||
|
||||
interface QuestionRow extends QueryResultRow, Omit<AppQuestion, 'created_at'> { created_at: Date }
|
||||
|
||||
interface SyncRow extends QueryResultRow {
|
||||
server_time: Date; couple_id: string | null; profile_updated_at: Date | null;
|
||||
partner_id: string | null; partner_updated_at: Date | null;
|
||||
match_count: number | null; new_match_count: number | null; latest_match_at: Date | null;
|
||||
pending_yours: number | null; pending_theirs: number | null; unread_total: number | null;
|
||||
enabled_packs_fingerprint: string | null;
|
||||
match_state_fingerprint: string | null; match_unread_fingerprint: string | null;
|
||||
streak_updated_at: Date | null;
|
||||
}
|
||||
|
||||
function question(row: QuestionRow): AppQuestion { return { ...row, created_at: row.created_at.toISOString() }; }
|
||||
|
||||
export class PostgresAppDataRepository implements AppDataRepository {
|
||||
private readonly pool: Pool;
|
||||
constructor(databaseUrl: string) { this.pool = new Pool({ connectionString: databaseUrl }); }
|
||||
private readonly ownsPool: boolean;
|
||||
constructor(connection: DatabaseConnection) { const resolved = resolvePool(connection); this.pool = resolved.pool; this.ownsPool = resolved.owned; }
|
||||
|
||||
private async visiblePack(userId: string, packId: string): Promise<PackContext> {
|
||||
const result = await this.pool.query<PackContext>(`
|
||||
@@ -189,5 +246,75 @@ export class PostgresAppDataRepository implements AppDataRepository {
|
||||
}
|
||||
return { strokes: row.strokes, revision: Number(row.revision), updated_at: row.updated_at.toISOString(), updated_by: row.updated_by };
|
||||
}
|
||||
async close(): Promise<void> { await this.pool.end(); }
|
||||
/**
|
||||
* One round trip, all scalar subqueries, nothing that grows with the couple's
|
||||
* history. An unpaired member's couple-scoped markers all collapse to zero or null
|
||||
* because `me.couple_id` is null, so the endpoint stays valid before pairing.
|
||||
*/
|
||||
async syncSummary(userId: string): Promise<SyncSummary> {
|
||||
const result = await this.pool.query<SyncRow>(`
|
||||
with me as (select id, couple_id, updated_at from profiles where id = $1),
|
||||
visible as (
|
||||
select m.id, m.is_new, m.created_at, m.match_type, m.response_summary from matches m, me
|
||||
where m.couple_id = me.couple_id
|
||||
and not exists (select 1 from match_archives a where a.match_id = m.id and a.user_id = me.id)
|
||||
),
|
||||
visible_unread as (
|
||||
select v.id as match_id, count(msg.id)::int as c
|
||||
from visible v
|
||||
left join messages msg on msg.match_id = v.id and msg.user_id <> (select id from me)
|
||||
and msg.read_at is null and msg.deleted_at is null
|
||||
and not exists (select 1 from message_deletions d where d.message_id = msg.id and d.user_id = (select id from me))
|
||||
group by v.id
|
||||
)
|
||||
select
|
||||
now() as server_time,
|
||||
me.couple_id,
|
||||
me.updated_at as profile_updated_at,
|
||||
(select p.id from profiles p where p.couple_id = me.couple_id and p.id <> me.id order by p.id limit 1) as partner_id,
|
||||
(select max(p.updated_at) from profiles p where p.couple_id = me.couple_id and p.id <> me.id) as partner_updated_at,
|
||||
(select count(*)::int from visible) as match_count,
|
||||
(select count(*)::int from visible where is_new) as new_match_count,
|
||||
(select max(created_at) from visible) as latest_match_at,
|
||||
(select count(*)::int from responses r join questions q on q.id = r.question_id
|
||||
where r.couple_id = me.couple_id and q.deleted_at is null and r.user_id <> me.id
|
||||
and not exists (select 1 from responses o
|
||||
where o.question_id = r.question_id and o.couple_id = me.couple_id and o.user_id = me.id)) as pending_yours,
|
||||
(select count(*)::int from responses r join questions q on q.id = r.question_id
|
||||
where r.couple_id = me.couple_id and q.deleted_at is null and r.user_id = me.id
|
||||
and not exists (select 1 from responses o
|
||||
where o.question_id = r.question_id and o.couple_id = me.couple_id and o.user_id <> me.id)) as pending_theirs,
|
||||
(select count(*)::int from messages msg join matches ma on ma.id = msg.match_id
|
||||
where ma.couple_id = me.couple_id and msg.user_id <> me.id and msg.read_at is null and msg.deleted_at is null
|
||||
and not exists (select 1 from message_deletions d where d.message_id = msg.id and d.user_id = me.id)) as unread_total,
|
||||
(select md5(coalesce(string_agg(cp.pack_id::text, ',' order by cp.pack_id), ''))
|
||||
from couple_packs cp where cp.couple_id = me.couple_id and cp.enabled) as enabled_packs_fingerprint,
|
||||
(select md5(coalesce(string_agg(v.id::text || ':' || v.match_type || ':' || coalesce(v.response_summary::text, ''), ',' order by v.id), ''))
|
||||
from visible v) as match_state_fingerprint,
|
||||
(select md5(coalesce(string_agg(vu.match_id::text || ':' || vu.c, ',' order by vu.match_id), ''))
|
||||
from visible_unread vu) as match_unread_fingerprint,
|
||||
(select cs.updated_at from couple_streaks cs where cs.couple_id = me.couple_id) as streak_updated_at
|
||||
from me`, [userId]);
|
||||
const row = result.rows[0];
|
||||
if (!row) throw new AppDataError('not_found', 'Profile not found', 404);
|
||||
return {
|
||||
server_time: row.server_time.toISOString(),
|
||||
couple_id: row.couple_id,
|
||||
profile_updated_at: row.profile_updated_at?.toISOString() ?? null,
|
||||
partner_id: row.partner_id,
|
||||
partner_updated_at: row.partner_updated_at?.toISOString() ?? null,
|
||||
match_count: row.match_count ?? 0,
|
||||
new_match_count: row.new_match_count ?? 0,
|
||||
latest_match_at: row.latest_match_at?.toISOString() ?? null,
|
||||
pending_yours: row.pending_yours ?? 0,
|
||||
pending_theirs: row.pending_theirs ?? 0,
|
||||
unread_total: row.unread_total ?? 0,
|
||||
enabled_packs_fingerprint: row.enabled_packs_fingerprint ?? '',
|
||||
match_state_fingerprint: row.match_state_fingerprint ?? '',
|
||||
match_unread_fingerprint: row.match_unread_fingerprint ?? '',
|
||||
streak_updated_at: row.streak_updated_at?.toISOString() ?? null,
|
||||
};
|
||||
}
|
||||
|
||||
async close(): Promise<void> { await closeResolvedPool(this.pool, this.ownsPool); }
|
||||
}
|
||||
@@ -44,6 +44,11 @@ export function registerAppDataRoutes(app: App, repository: AppDataRepository):
|
||||
const result = await run(() => repository.markMediaViewed(c.get('identity').id, id.data, body.data.expires_at));
|
||||
return result.ok ? c.json(result.value) : c.json(error(result.cause.code, result.cause.message), result.cause.status);
|
||||
});
|
||||
// The client polls this instead of re-fetching every domain; see SyncSummary.
|
||||
app.get('/v1/me/sync', async c => {
|
||||
const result = await run(() => repository.syncSummary(c.get('identity').id));
|
||||
return result.ok ? c.json(result.value) : c.json(error(result.cause.code, result.cause.message), result.cause.status);
|
||||
});
|
||||
app.get('/v1/me/nudge-status', async c => c.json(await repository.nudgeStatus(c.get('identity').id)));
|
||||
app.post('/v1/me/nudge', async c => {
|
||||
const result = await run(() => repository.sendNudge(c.get('identity').id));
|
||||
|
||||
@@ -1,4 +1,5 @@
|
||||
import { Pool, type PoolClient, type QueryResultRow } from 'pg';
|
||||
import { closeResolvedPool, resolvePool, type DatabaseConnection } from '../../db/pool.js';
|
||||
import { BillingError, type RedemptionResult, type RevenueCatWebhook, type SubscriptionStatus } from './types.js';
|
||||
|
||||
interface ProfileRecord extends QueryResultRow {
|
||||
@@ -43,9 +44,12 @@ async function transaction<T>(pool: Pool, operation: (client: PoolClient) => Pro
|
||||
|
||||
export class PostgresBillingRepository implements BillingRepository {
|
||||
private readonly pool: Pool;
|
||||
private readonly ownsPool: boolean;
|
||||
|
||||
constructor(databaseUrl: string) {
|
||||
this.pool = new Pool({ connectionString: databaseUrl });
|
||||
constructor(connection: DatabaseConnection) {
|
||||
const resolved = resolvePool(connection);
|
||||
this.pool = resolved.pool;
|
||||
this.ownsPool = resolved.owned;
|
||||
}
|
||||
|
||||
async processRevenueCatEvent(
|
||||
@@ -207,6 +211,6 @@ export class PostgresBillingRepository implements BillingRepository {
|
||||
}
|
||||
|
||||
async close(): Promise<void> {
|
||||
await this.pool.end();
|
||||
await closeResolvedPool(this.pool, this.ownsPool);
|
||||
}
|
||||
}
|
||||
@@ -1,4 +1,5 @@
|
||||
import { Pool, type QueryResultRow } from 'pg';
|
||||
import { closeResolvedPool, resolvePool, type DatabaseConnection } from '../../db/pool.js';
|
||||
|
||||
export type ChatMediaType = 'image' | 'video' | null;
|
||||
export type ReportReason = 'harassment' | 'spam' | 'inappropriate_content' | 'other';
|
||||
@@ -45,8 +46,37 @@ export class ChatError extends Error {
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* The reply to a message poll. A request without a cursor is a full snapshot; a
|
||||
* request with one carries only what changed, so a quiet chat costs an empty
|
||||
* array rather than the whole backlog.
|
||||
*/
|
||||
export interface MessagesPage {
|
||||
/** Newest first. Messages created or changed since the cursor, or the snapshot. */
|
||||
messages: ChatMessage[];
|
||||
/** Messages the caller deleted for themselves since the cursor. */
|
||||
removed_ids: string[];
|
||||
/** Cursor to send as `since` on the next poll. */
|
||||
server_time: string;
|
||||
/** True when `messages` is a complete snapshot the client may replace its cache with. */
|
||||
complete: boolean;
|
||||
/** Present only when the caller asked for it, folding the typing poll into this one. */
|
||||
typing?: PartnerTyping;
|
||||
}
|
||||
|
||||
export interface PartnerTyping {
|
||||
typing: boolean;
|
||||
expires_at: string | null;
|
||||
}
|
||||
|
||||
export interface ListMessagesOptions {
|
||||
/** ISO timestamp from a previous `server_time`. Omit for a full snapshot. */
|
||||
since?: string;
|
||||
includeTyping?: boolean;
|
||||
}
|
||||
|
||||
export interface ChatRepository {
|
||||
listMessages(userId: string, matchId: string): Promise<ChatMessage[]>;
|
||||
listMessages(userId: string, matchId: string, options?: ListMessagesOptions): Promise<MessagesPage>;
|
||||
sendText(userId: string, matchId: string, content: string): Promise<ChatMessage>;
|
||||
unreadCounts(userId: string): Promise<{ total: number; by_match: Record<string, number> }>;
|
||||
markMatchRead(userId: string, matchId: string): Promise<{ updated: number; read_at: string }>;
|
||||
@@ -55,13 +85,21 @@ export interface ChatRepository {
|
||||
deleteForEveryone(userId: string, messageId: string): Promise<ChatMessage>;
|
||||
report(userId: string, messageId: string, reason: ReportReason): Promise<void>;
|
||||
setTyping(userId: string, matchId: string, ttlMs: number): Promise<void>;
|
||||
getPartnerTyping(userId: string, matchId: string): Promise<{ typing: boolean; expires_at: string | null }>;
|
||||
getPartnerTyping(userId: string, matchId: string): Promise<PartnerTyping>;
|
||||
close(): Promise<void>;
|
||||
}
|
||||
|
||||
// `media_expired` is derived rather than read. Nothing ever writes the stored
|
||||
// column — it is inserted `false` and never updated — so a video that reached
|
||||
// `media_expires_at` still arrived flagged as playable, while the media itself
|
||||
// was already refused (see `accessible` in the media repository, which treats
|
||||
// `media_expired or media_expires_at <= now()` as gone). Deriving it here is the
|
||||
// same rule the media side already applies, and it is what makes the client's
|
||||
// "Video expired" state reachable at all.
|
||||
const columns = `m.id, m.match_id, m.user_id, m.content, m.created_at, m.read_at,
|
||||
m.delivered_at, m.deleted_at, m.media_path, m.media_type, m.media_expires_at,
|
||||
m.media_expired, m.media_viewed_at, m.version, m.encrypted_content,
|
||||
(m.media_expired or (m.media_expires_at is not null and m.media_expires_at <= now())) as media_expired,
|
||||
m.media_viewed_at, m.version, m.encrypted_content,
|
||||
m.encryption_iv, m.keys_metadata, m.moderation_status, m.flag_reason, m.category`;
|
||||
|
||||
function iso(value: Date | null): string | null {
|
||||
@@ -82,9 +120,12 @@ function toMessage(row: MessageRecord): ChatMessage {
|
||||
|
||||
export class PostgresChatRepository implements ChatRepository {
|
||||
private readonly pool: Pool;
|
||||
private readonly ownsPool: boolean;
|
||||
|
||||
constructor(databaseUrl: string) {
|
||||
this.pool = new Pool({ connectionString: databaseUrl });
|
||||
constructor(connection: DatabaseConnection) {
|
||||
const resolved = resolvePool(connection);
|
||||
this.pool = resolved.pool;
|
||||
this.ownsPool = resolved.owned;
|
||||
}
|
||||
|
||||
private async assertMatchAccess(userId: string, matchId: string): Promise<void> {
|
||||
@@ -112,20 +153,68 @@ export class PostgresChatRepository implements ChatRepository {
|
||||
return row;
|
||||
}
|
||||
|
||||
async listMessages(userId: string, matchId: string): Promise<ChatMessage[]> {
|
||||
/**
|
||||
* Without `since` this is the previous behaviour: the newest 500 visible messages.
|
||||
* With `since` it returns only rows whose visible state moved after that instant —
|
||||
* new messages, existing ones whose delivery, read, deletion or media-view
|
||||
* timestamp advanced, and videos whose media expired inside the window — plus the
|
||||
* ids the caller has since deleted for themselves.
|
||||
*
|
||||
* Moderation metadata (`moderation_status`, `flag_reason`, `category`) is not
|
||||
* covered: the classifier and the admin tool rewrite those columns without
|
||||
* advancing any timestamp, so a delta cannot see them move. No client renders
|
||||
* them today, so nothing user-visible goes stale; giving them a cursor would
|
||||
* mean adding a column to `messages`.
|
||||
* Rows are returned whole rather than as patches so the client merges by id and
|
||||
* needs no second shape to interpret.
|
||||
*
|
||||
* The returned cursor is deliberately a couple of seconds behind the server clock:
|
||||
* a transaction that started before this read but committed after it would
|
||||
* otherwise fall into the gap between two polls. The overlap re-delivers a handful
|
||||
* of rows the client already has, which its merge-by-id absorbs.
|
||||
*/
|
||||
async listMessages(userId: string, matchId: string, options: ListMessagesOptions = {}): Promise<MessagesPage> {
|
||||
await this.assertMatchAccess(userId, matchId);
|
||||
const result = await this.pool.query<MessageRecord>(
|
||||
`select ${columns}
|
||||
from messages m
|
||||
where m.match_id = $1
|
||||
and not exists (
|
||||
select 1 from message_deletions d where d.message_id = m.id and d.user_id = $2
|
||||
)
|
||||
order by m.created_at desc, m.id desc
|
||||
limit 500`,
|
||||
[matchId, userId],
|
||||
);
|
||||
return result.rows.map(toMessage);
|
||||
const since = options.since ?? null;
|
||||
const visible = `not exists (select 1 from message_deletions d where d.message_id = m.id and d.user_id = $2)`;
|
||||
// Expiry is the one visible change no row write announces: the clock crosses
|
||||
// `media_expires_at` and nothing is updated, so no timestamp moves past the
|
||||
// cursor. Comparing the deadline against the window instead emits the row in
|
||||
// the single poll whose `(since, now()]` contains it, which is exactly when
|
||||
// the derived `media_expired` above flips.
|
||||
const changed = `(m.created_at > $3 or m.read_at > $3 or m.delivered_at > $3
|
||||
or m.deleted_at > $3 or m.media_viewed_at > $3
|
||||
or (m.media_expires_at > $3 and m.media_expires_at <= now()))`;
|
||||
|
||||
const [clock, messages, removed, typing] = await Promise.all([
|
||||
this.pool.query<{ cursor: Date }>(`select now() - interval '2 seconds' as cursor`),
|
||||
this.pool.query<MessageRecord>(
|
||||
`select ${columns}
|
||||
from messages m
|
||||
where m.match_id = $1 and ${visible}${since === null ? '' : ` and ${changed}`}
|
||||
order by m.created_at desc, m.id desc
|
||||
limit 500`,
|
||||
since === null ? [matchId, userId] : [matchId, userId, since],
|
||||
),
|
||||
since === null
|
||||
? null
|
||||
: this.pool.query<{ message_id: string }>(
|
||||
`select d.message_id
|
||||
from message_deletions d
|
||||
join messages m on m.id = d.message_id
|
||||
where m.match_id = $1 and d.user_id = $2 and d.deleted_at > $3`,
|
||||
[matchId, userId, since],
|
||||
),
|
||||
options.includeTyping ? this.partnerTyping(userId, matchId) : null,
|
||||
]);
|
||||
|
||||
return {
|
||||
messages: messages.rows.map(toMessage),
|
||||
removed_ids: removed?.rows.map((row) => row.message_id) ?? [],
|
||||
server_time: clock.rows[0]!.cursor.toISOString(),
|
||||
complete: since === null,
|
||||
...(typing ? { typing } : {}),
|
||||
};
|
||||
}
|
||||
|
||||
async sendText(userId: string, matchId: string, content: string): Promise<ChatMessage> {
|
||||
@@ -226,8 +315,8 @@ export class PostgresChatRepository implements ChatRepository {
|
||||
);
|
||||
}
|
||||
|
||||
async getPartnerTyping(userId: string, matchId: string): Promise<{ typing: boolean; expires_at: string | null }> {
|
||||
await this.assertMatchAccess(userId, matchId);
|
||||
/** Access already asserted by the caller, so the message poll can fold this in for free. */
|
||||
private async partnerTyping(userId: string, matchId: string): Promise<PartnerTyping> {
|
||||
const result = await this.pool.query<{ expires_at: Date }>(
|
||||
`select expires_at from chat_typing_states
|
||||
where match_id = $1 and user_id <> $2 and expires_at > now()
|
||||
@@ -238,8 +327,13 @@ export class PostgresChatRepository implements ChatRepository {
|
||||
return { typing: Boolean(expiresAt), expires_at: iso(expiresAt) };
|
||||
}
|
||||
|
||||
async getPartnerTyping(userId: string, matchId: string): Promise<PartnerTyping> {
|
||||
await this.assertMatchAccess(userId, matchId);
|
||||
return this.partnerTyping(userId, matchId);
|
||||
}
|
||||
|
||||
async close(): Promise<void> {
|
||||
await this.pool.end();
|
||||
await closeResolvedPool(this.pool, this.ownsPool);
|
||||
}
|
||||
}
|
||||
|
||||
@@ -8,6 +8,7 @@ type AuthenticatedApp = Hono<{ Variables: { identity: AuthIdentity } }>;
|
||||
const id = z.string().uuid();
|
||||
const sendBody = z.object({ content: z.string().trim().min(1).max(10_000) }).strict();
|
||||
const reportBody = z.object({ reason: z.enum(['harassment', 'spam', 'inappropriate_content', 'other']) }).strict();
|
||||
const since = z.string().datetime({ offset: true });
|
||||
|
||||
function error(code: string, message: string): ApiErrorResponse {
|
||||
return { error: { code, message } };
|
||||
@@ -25,11 +26,20 @@ async function result<T>(operation: () => Promise<T>) {
|
||||
export function registerChatRoutes(app: AuthenticatedApp, repository: ChatRepository): void {
|
||||
app.get('/v1/chat/unread', async (c) => c.json(await repository.unreadCounts(c.get('identity').id)));
|
||||
|
||||
// `since` turns this into a delta poll and `typing=true` folds the typing check
|
||||
// into the same request. Both are optional: a call with neither still answers with
|
||||
// the full snapshot under the `messages` key it always used.
|
||||
app.get('/v1/matches/:matchId/messages', async (c) => {
|
||||
const parsedId = id.safeParse(c.req.param('matchId'));
|
||||
if (!parsedId.success) return c.json(error('invalid_match_id', 'A valid match ID is required'), 400);
|
||||
const response = await result(() => repository.listMessages(c.get('identity').id, parsedId.data));
|
||||
return response.ok ? c.json({ messages: response.value }) : c.json(error(response.cause.code, response.cause.message), response.cause.status);
|
||||
const rawSince = c.req.query('since');
|
||||
const parsedSince = rawSince === undefined ? undefined : since.safeParse(rawSince);
|
||||
if (parsedSince && !parsedSince.success) return c.json(error('invalid_since', 'since must be an ISO-8601 timestamp'), 400);
|
||||
const response = await result(() => repository.listMessages(c.get('identity').id, parsedId.data, {
|
||||
...(parsedSince?.success ? { since: parsedSince.data } : {}),
|
||||
includeTyping: c.req.query('typing') === 'true',
|
||||
}));
|
||||
return response.ok ? c.json(response.value) : c.json(error(response.cause.code, response.cause.message), response.cause.status);
|
||||
});
|
||||
|
||||
app.post('/v1/matches/:matchId/messages', async (c) => {
|
||||
|
||||
@@ -1,5 +1,6 @@
|
||||
import type { Couple, Profile } from '@sauci/shared';
|
||||
import { Pool, type PoolClient, type QueryResultRow } from 'pg';
|
||||
import { closeResolvedPool, resolvePool, type DatabaseConnection } from '../../db/pool.js';
|
||||
import { CoupleError, type CoupleStateResponse } from './types.js';
|
||||
|
||||
interface ProfileRecord extends QueryResultRow {
|
||||
@@ -93,9 +94,12 @@ async function transaction<T>(pool: Pool, work: (client: PoolClient) => Promise<
|
||||
|
||||
export class PostgresCoupleRepository implements CoupleRepository {
|
||||
private readonly pool: Pool;
|
||||
private readonly ownsPool: boolean;
|
||||
|
||||
constructor(databaseUrl: string) {
|
||||
this.pool = new Pool({ connectionString: databaseUrl });
|
||||
constructor(connection: DatabaseConnection) {
|
||||
const resolved = resolvePool(connection);
|
||||
this.pool = resolved.pool;
|
||||
this.ownsPool = resolved.owned;
|
||||
}
|
||||
|
||||
async getState(userId: string): Promise<CoupleStateResponse> {
|
||||
@@ -199,6 +203,6 @@ export class PostgresCoupleRepository implements CoupleRepository {
|
||||
}
|
||||
|
||||
async close(): Promise<void> {
|
||||
await this.pool.end();
|
||||
await closeResolvedPool(this.pool, this.ownsPool);
|
||||
}
|
||||
}
|
||||
@@ -1,4 +1,5 @@
|
||||
import { Pool, type PoolClient, type QueryResultRow } from 'pg';
|
||||
import { closeResolvedPool, resolvePool, type DatabaseConnection } from '../../db/pool.js';
|
||||
|
||||
export type DareStatus =
|
||||
| 'pending' | 'active' | 'submitted' | 'completed' | 'expired' | 'declined' | 'cancelled';
|
||||
@@ -181,9 +182,12 @@ function toSentDare(row: SentDareRecord, viewerId: string): SentDare {
|
||||
|
||||
export class PostgresDaresRepository implements DaresRepository {
|
||||
private readonly pool: Pool;
|
||||
private readonly ownsPool: boolean;
|
||||
|
||||
constructor(databaseUrl: string) {
|
||||
this.pool = new Pool({ connectionString: databaseUrl });
|
||||
constructor(connection: DatabaseConnection) {
|
||||
const resolved = resolvePool(connection);
|
||||
this.pool = resolved.pool;
|
||||
this.ownsPool = resolved.owned;
|
||||
}
|
||||
|
||||
/**
|
||||
@@ -475,6 +479,6 @@ export class PostgresDaresRepository implements DaresRepository {
|
||||
}
|
||||
|
||||
async close(): Promise<void> {
|
||||
await this.pool.end();
|
||||
await closeResolvedPool(this.pool, this.ownsPool);
|
||||
}
|
||||
}
|
||||
@@ -1,5 +1,6 @@
|
||||
import { randomUUID } from 'node:crypto';
|
||||
import { Pool, type PoolClient, type QueryResultRow } from 'pg';
|
||||
import { closeResolvedPool, resolvePool, type DatabaseConnection } from '../../db/pool.js';
|
||||
import type { MediaKind, MediaObject, MediaUploadContext } from './types.js';
|
||||
|
||||
interface MediaRow extends QueryResultRow {
|
||||
@@ -28,7 +29,8 @@ function object(row: MediaRow): MediaObject {
|
||||
|
||||
export class PostgresMediaRepository implements MediaRepository {
|
||||
private readonly pool: Pool;
|
||||
constructor(databaseUrl: string) { this.pool = new Pool({ connectionString: databaseUrl }); }
|
||||
private readonly ownsPool: boolean;
|
||||
constructor(connection: DatabaseConnection) { const resolved = resolvePool(connection); this.pool = resolved.pool; this.ownsPool = resolved.owned; }
|
||||
|
||||
async create(userId: string, kind: MediaKind, storageKey: string, mimeType: string, byteSize: number, context: MediaUploadContext) {
|
||||
const client = await this.pool.connect();
|
||||
@@ -102,5 +104,5 @@ export class PostgresMediaRepository implements MediaRepository {
|
||||
async acknowledgeDeletion(storageKey: string): Promise<void> {
|
||||
await this.pool.query('delete from media_deletion_queue where storage_key=$1',[storageKey]);
|
||||
}
|
||||
async close() { await this.pool.end(); }
|
||||
async close(): Promise<void> { await closeResolvedPool(this.pool, this.ownsPool); }
|
||||
}
|
||||
@@ -1,4 +1,5 @@
|
||||
import { Pool, type PoolClient, type QueryResultRow } from 'pg';
|
||||
import { closeResolvedPool, resolvePool, type DatabaseConnection } from '../../db/pool.js';
|
||||
import type { ClassifierRuntimeConfig, MessageForClassification, OperationItem, ProducerSummary } from './types.js';
|
||||
|
||||
interface OutboxRow extends QueryResultRow {
|
||||
@@ -31,7 +32,8 @@ async function transaction<T>(pool: Pool, work: (client: PoolClient) => Promise<
|
||||
|
||||
export class PostgresOperationsRepository implements OperationsRepository {
|
||||
private readonly pool: Pool;
|
||||
constructor(databaseUrl: string) { this.pool = new Pool({ connectionString: databaseUrl }); }
|
||||
private readonly ownsPool: boolean;
|
||||
constructor(connection: DatabaseConnection) { const resolved = resolvePool(connection); this.pool = resolved.pool; this.ownsPool = resolved.owned; }
|
||||
|
||||
async produce(now: Date, limit=100): Promise<ProducerSummary> {
|
||||
return transaction(this.pool, async (client) => {
|
||||
@@ -312,7 +314,7 @@ export class PostgresOperationsRepository implements OperationsRepository {
|
||||
async classify(id:string,status:'safe'|'flagged',reason:string|null,category:string) {
|
||||
await this.pool.query('update messages set moderation_status=$2,flag_reason=$3,category=$4 where id=$1',[id,status,reason,category]);
|
||||
}
|
||||
async close() { await this.pool.end(); }
|
||||
async close(): Promise<void> { await closeResolvedPool(this.pool, this.ownsPool); }
|
||||
}
|
||||
|
||||
function streakCopy(days:number) {
|
||||
|
||||
@@ -1,4 +1,5 @@
|
||||
import { Pool } from 'pg';
|
||||
import { closeResolvedPool, resolvePool, type DatabaseConnection } from '../../db/pool.js';
|
||||
|
||||
export interface PackCategory {
|
||||
id: string;
|
||||
@@ -69,9 +70,12 @@ interface ProfilePreferencesRow {
|
||||
|
||||
export class PostgresPacksRepository implements PacksRepository {
|
||||
private readonly pool: Pool;
|
||||
private readonly ownsPool: boolean;
|
||||
|
||||
constructor(databaseUrl: string) {
|
||||
this.pool = new Pool({ connectionString: databaseUrl });
|
||||
constructor(connection: DatabaseConnection) {
|
||||
const resolved = resolvePool(connection);
|
||||
this.pool = resolved.pool;
|
||||
this.ownsPool = resolved.owned;
|
||||
}
|
||||
|
||||
private async profile(userId: string): Promise<ProfilePreferencesRow> {
|
||||
@@ -215,6 +219,6 @@ export class PostgresPacksRepository implements PacksRepository {
|
||||
}
|
||||
|
||||
async close(): Promise<void> {
|
||||
await this.pool.end();
|
||||
await closeResolvedPool(this.pool, this.ownsPool);
|
||||
}
|
||||
}
|
||||
@@ -1,5 +1,6 @@
|
||||
import { randomUUID } from 'node:crypto';
|
||||
import { Pool, type QueryResultRow } from 'pg';
|
||||
import { closeResolvedPool, resolvePool, type DatabaseConnection } from '../../db/pool.js';
|
||||
import type {
|
||||
FeedbackSubmission,
|
||||
NotificationPreferenceKey,
|
||||
@@ -80,9 +81,12 @@ function assertChanged(rowCount: number | null): void {
|
||||
|
||||
export class PostgresProfileSettingsRepository implements ProfileSettingsRepository {
|
||||
private readonly pool: Pool;
|
||||
private readonly ownsPool: boolean;
|
||||
|
||||
constructor(databaseUrl: string) {
|
||||
this.pool = new Pool({ connectionString: databaseUrl });
|
||||
constructor(connection: DatabaseConnection) {
|
||||
const resolved = resolvePool(connection);
|
||||
this.pool = resolved.pool;
|
||||
this.ownsPool = resolved.owned;
|
||||
}
|
||||
|
||||
async updateProfile(userId: string, update: ProfileUpdate): Promise<void> {
|
||||
@@ -146,6 +150,6 @@ export class PostgresProfileSettingsRepository implements ProfileSettingsReposit
|
||||
}
|
||||
|
||||
async close(): Promise<void> {
|
||||
await this.pool.end();
|
||||
await closeResolvedPool(this.pool, this.ownsPool);
|
||||
}
|
||||
}
|
||||
+20
-26
@@ -2,6 +2,7 @@ import { serve } from '@hono/node-server';
|
||||
import { createApp } from './app.js';
|
||||
import { createAuthVerifier } from './auth.js';
|
||||
import { loadConfig } from './config.js';
|
||||
import { createDatabasePool } from './db/pool.js';
|
||||
import { PostgresRepository } from './db/repository.js';
|
||||
import { PostgresAppDataRepository } from './domains/app-data/repository.js';
|
||||
import { AdminRequestAuth } from './domains/admin/auth.js';
|
||||
@@ -26,30 +27,34 @@ import { FilesystemMediaStorage, PostgresMediaRepository } from './domains/media
|
||||
|
||||
const config = loadConfig();
|
||||
const authVerifier = createAuthVerifier(config);
|
||||
const repository = new PostgresRepository(config.databaseUrl);
|
||||
const coupleRepository = new PostgresCoupleRepository(config.databaseUrl);
|
||||
const packsRepository = new PostgresPacksRepository(config.databaseUrl);
|
||||
const answersRepository = new PostgresAnswersRepository(config.databaseUrl);
|
||||
const chatRepository = new PostgresChatRepository(config.databaseUrl);
|
||||
const daresRepository = new PostgresDaresRepository(config.databaseUrl);
|
||||
const profileSettingsRepository = new PostgresProfileSettingsRepository(config.databaseUrl);
|
||||
const accountOperationsRepository = new PostgresAccountOperationsRepository(config.databaseUrl);
|
||||
// One deliberately bounded pool for the whole process. Repositories used to open
|
||||
// a default-sized pool each, so a single API instance could hold twelve times the
|
||||
// connections a self-hosted PostgreSQL was sized for.
|
||||
const pool = createDatabasePool(config.databaseUrl, config.databasePool);
|
||||
const repository = new PostgresRepository(pool);
|
||||
const coupleRepository = new PostgresCoupleRepository(pool);
|
||||
const packsRepository = new PostgresPacksRepository(pool);
|
||||
const answersRepository = new PostgresAnswersRepository(pool);
|
||||
const chatRepository = new PostgresChatRepository(pool);
|
||||
const daresRepository = new PostgresDaresRepository(pool);
|
||||
const profileSettingsRepository = new PostgresProfileSettingsRepository(pool);
|
||||
const accountOperationsRepository = new PostgresAccountOperationsRepository(pool);
|
||||
const accountOperationsService = new AccountOperationsService(
|
||||
accountOperationsRepository,
|
||||
new SupabaseAuthAdminClient(config.authIssuer, config.supabaseAuthServiceRoleKey),
|
||||
new HttpRevenueCatClient(config.revenueCatApiKey, config.revenueCatEntitlementId),
|
||||
new ExpoPartnerNotifier(),
|
||||
);
|
||||
const mediaRepository = new PostgresMediaRepository(config.databaseUrl);
|
||||
const mediaRepository = new PostgresMediaRepository(pool);
|
||||
const mediaStorage = new FilesystemMediaStorage(
|
||||
config.mediaRoot,
|
||||
config.mediaSigningSecret,
|
||||
config.mediaPublicBaseUrl,
|
||||
);
|
||||
const appDataRepository = new PostgresAppDataRepository(config.databaseUrl);
|
||||
const billingRepository = new PostgresBillingRepository(config.databaseUrl);
|
||||
const appDataRepository = new PostgresAppDataRepository(pool);
|
||||
const billingRepository = new PostgresBillingRepository(pool);
|
||||
const billingService = new BillingService(billingRepository, config.revenueCatWebhookSecret);
|
||||
const adminRepository = new PostgresAdminRepository(config.databaseUrl, {
|
||||
const adminRepository = new PostgresAdminRepository(pool, {
|
||||
adminPrivateKeyJwk: config.adminPrivateKeyJwk,
|
||||
mediaRoot: config.mediaRoot,
|
||||
authDirectory: new SupabaseAdminAuthDirectory(config.authIssuer, config.supabaseAuthServiceRoleKey),
|
||||
@@ -94,20 +99,9 @@ async function shutdown(signal: string): Promise<void> {
|
||||
|
||||
server.close(async () => {
|
||||
try {
|
||||
await Promise.all([
|
||||
repository.close(),
|
||||
coupleRepository.close(),
|
||||
packsRepository.close(),
|
||||
answersRepository.close(),
|
||||
chatRepository.close(),
|
||||
daresRepository.close(),
|
||||
profileSettingsRepository.close(),
|
||||
accountOperationsRepository.close(),
|
||||
mediaRepository.close(),
|
||||
appDataRepository.close(),
|
||||
billingRepository.close(),
|
||||
adminRepository.close(),
|
||||
]);
|
||||
// Repositories share the process pool and no longer own a connection each,
|
||||
// so ending the pool once is the whole of the database shutdown.
|
||||
await pool.end();
|
||||
process.exit(0);
|
||||
} catch (error) {
|
||||
console.error('Failed to close database connections', error);
|
||||
|
||||
@@ -6,7 +6,7 @@ import { PostgresAnswersRepository } from '../src/domains/answers/repository.js'
|
||||
const url=process.env.DATABASE_URL;const local=url?['127.0.0.1','localhost','::1'].includes(new URL(url).hostname):false;if(url&&!local)throw new Error('Answer integration tests only permit localhost');
|
||||
describe.skipIf(!url||!local)('answers + PostgreSQL',()=>{const admin=new Pool({connectionString:url});const schema=`answers_${randomUUID().replaceAll('-','')}`;let pool:Pool;let repo:PostgresAnswersRepository;
|
||||
const u1='11111111-1111-4111-8111-111111111111',u2='22222222-2222-4222-8222-222222222222',outsider='33333333-3333-4333-8333-333333333333',couple='44444444-4444-4444-8444-444444444444',other='55555555-5555-4555-8555-555555555555',pack='66666666-6666-4666-8666-666666666666',q='77777777-7777-4777-8777-777777777777',concurrentQ='88888888-8888-4888-8888-888888888888',femaleOnlyQ='99999999-9999-4999-8999-999999999999',limitQ1='aaaaaaaa-aaaa-4aaa-8aaa-aaaaaaaaaaaa',limitQ2='bbbbbbbb-bbbb-4bbb-8bbb-bbbbbbbbbbbb',catchUpQ='cccccccc-cccc-4ccc-8ccc-cccccccccccc';
|
||||
beforeAll(async()=>{await admin.query(`create schema "${schema}"`);const isolated=new URL(url!);isolated.searchParams.set('options',`-c search_path=${schema}`);pool=new Pool({connectionString:isolated.toString()});repo=new PostgresAnswersRepository(isolated.toString());for(const name of ['0000_identity_and_feature_interests.sql','0001_couples.sql','0002_packs_catalog_progress.sql','0003_answers_matches.sql','0014_daily_limit_local_reset.sql','0015_daily_limit_response_index.sql','0016_couple_streak_locality.sql']){const sql=await readFile(new URL(`../drizzle/${name}`,import.meta.url),'utf8');for(const s of sql.split('--> statement-breakpoint'))if(s.trim())await pool.query(s);}await pool.query('insert into couples(id,invite_code) values($1,$2),($3,$4)',[couple,'ANSWERS1',other,'ANSWERS2']);await pool.query("insert into profiles(id,couple_id,gender,max_intensity) values($1,$4,'female',5),($2,$4,'male',3),($3,$5,'male',5)",[u1,u2,outsider,couple,other]);await pool.query("insert into question_packs(id,name) values($1,'Core')",[pack]);await pool.query("insert into questions(id,pack_id,text,intensity,target_user_genders) values($1,$2,'Question',3,array['female']),($3,$2,'Concurrent',2,null),($4,$2,'Female first',2,array['female']),($5,$2,'Limit one',2,null),($6,$2,'Limit two',2,null),($7,$2,'Catch up',2,null)",[q,pack,concurrentQ,femaleOnlyQ,limitQ1,limitQ2,catchUpQ]);await pool.query('insert into couple_packs(couple_id,pack_id,enabled) values($1,$2,true)',[couple,pack]);});afterAll(async()=>{await repo.close();await pool.end();await admin.query(`drop schema "${schema}" cascade`);await admin.end();});
|
||||
beforeAll(async()=>{await admin.query(`create schema "${schema}"`);const isolated=new URL(url!);isolated.searchParams.set('options',`-c search_path=${schema}`);pool=new Pool({connectionString:isolated.toString()});repo=new PostgresAnswersRepository(isolated.toString());for(const name of ['0000_identity_and_feature_interests.sql','0001_couples.sql','0002_packs_catalog_progress.sql','0003_answers_matches.sql','0004_chat.sql','0014_daily_limit_local_reset.sql','0015_daily_limit_response_index.sql','0016_couple_streak_locality.sql']){const sql=await readFile(new URL(`../drizzle/${name}`,import.meta.url),'utf8');for(const s of sql.split('--> statement-breakpoint'))if(s.trim())await pool.query(s);}await pool.query('insert into couples(id,invite_code) values($1,$2),($3,$4)',[couple,'ANSWERS1',other,'ANSWERS2']);await pool.query("insert into profiles(id,couple_id,gender,max_intensity) values($1,$4,'female',5),($2,$4,'male',3),($3,$5,'male',5)",[u1,u2,outsider,couple,other]);await pool.query("insert into question_packs(id,name) values($1,'Core')",[pack]);await pool.query("insert into questions(id,pack_id,text,intensity,target_user_genders) values($1,$2,'Question',3,array['female']),($3,$2,'Concurrent',2,null),($4,$2,'Female first',2,array['female']),($5,$2,'Limit one',2,null),($6,$2,'Limit two',2,null),($7,$2,'Catch up',2,null)",[q,pack,concurrentQ,femaleOnlyQ,limitQ1,limitQ2,catchUpQ]);await pool.query('insert into couple_packs(couple_id,pack_id,enabled) values($1,$2,true)',[couple,pack]);});afterAll(async()=>{await repo.close();await pool.end();await admin.query(`drop schema "${schema}" cascade`);await admin.end();});
|
||||
it('targets recommendation by gender and excludes completed responses',async()=>{expect((await repo.recommended(u1)).map(x=>x.id)).toContain(q);expect((await repo.recommended(u2)).map(x=>x.id)).not.toContain(q);await repo.submit(u1,{questionId:q,answer:'yes'});expect((await repo.recommended(u1)).map(x=>x.id)).not.toContain(q);});
|
||||
it('rejects direct answers that bypass target-gender eligibility',async()=>{await expect(repo.submit(u2,{questionId:femaleOnlyQ,answer:'yes'})).rejects.toMatchObject({code:'question_not_eligible'});});
|
||||
it('serializes concurrent partner submissions and still creates one match',async()=>{await Promise.all([repo.submit(u1,{questionId:concurrentQ,answer:'yes'}),repo.submit(u2,{questionId:concurrentQ,answer:'yes'})]);const counts=await pool.query('select count(*)::int count from matches where question_id=$1',[concurrentQ]);expect(counts.rows[0].count).toBe(1);});
|
||||
@@ -18,6 +18,71 @@ it('answers the streak from the caller side rather than by stored position',asyn
|
||||
expect(await repo.streak(u1)).toMatchObject({you_answered_today:true,partner_answered_today:false,partner_name:'Alex',current_streak:0});
|
||||
expect(await repo.streak(u2)).toMatchObject({you_answered_today:false,partner_answered_today:true,partner_name:'Sam',current_streak:0});});
|
||||
|
||||
describe('match pagination',()=>{
|
||||
const paged=Array.from({length:5},(_,index)=>`d0000000-0000-4000-8000-00000000000${index}`);
|
||||
const pagedQuestions=Array.from({length:5},(_,index)=>`f0000000-0000-4000-8000-00000000000${index}`);
|
||||
const otherMatch='e0000000-0000-4000-8000-000000000000';
|
||||
beforeAll(async()=>{
|
||||
await pool.query('delete from match_archives');await pool.query('delete from messages');await pool.query('delete from matches');
|
||||
// One question per match: a couple may only match a given question once.
|
||||
for(const[index,id]of pagedQuestions.entries())await pool.query("insert into questions(id,pack_id,text,intensity) values($1,$2,$3,2)",[id,pack,`Paged ${index}`]);
|
||||
// Distinct creation times so the ordering assertion is about the query, not tie-breaking.
|
||||
for(const[index,id]of paged.entries())await pool.query("insert into matches(id,couple_id,question_id,match_type,created_at) values($1,$2,$3,'yes_yes',now()-($4::text||' minutes')::interval)",[id,couple,pagedQuestions[index],String(index)]);
|
||||
// A match belonging to another couple must never appear, whatever the page.
|
||||
await pool.query("insert into matches(id,couple_id,question_id,match_type) values($1,$2,$3,'yes_yes')",[otherMatch,other,q]);
|
||||
// Two unread from the partner on the oldest match; one message of our own, which is never unread.
|
||||
await pool.query("insert into messages(match_id,user_id,content) values($1,$2,'a'),($1,$2,'b'),($1,$3,'mine')",[paged[4],u2,u1]);
|
||||
});
|
||||
|
||||
it('counts once on the first page, never again, and still reports hasMore',async()=>{
|
||||
const first=await repo.matches(u1,0,2,false);
|
||||
expect(first.totalCount).toBe(5);
|
||||
expect(first.hasMore).toBe(true);
|
||||
expect(first.matches).toHaveLength(2);
|
||||
|
||||
// Later pages skip the count entirely: the client already has it and paging
|
||||
// is decided by hasMore, which comes from reading limit + 1 rows.
|
||||
const second=await repo.matches(u1,1,2,false);
|
||||
expect(second.totalCount).toBeNull();
|
||||
expect(second.hasMore).toBe(true);
|
||||
const last=await repo.matches(u1,2,2,false);
|
||||
expect(last.totalCount).toBeNull();
|
||||
expect(last.hasMore).toBe(false);
|
||||
expect(last.matches).toHaveLength(1);
|
||||
|
||||
// An exactly-full final page must not claim another page exists.
|
||||
expect(await repo.matches(u1,0,5,false)).toMatchObject({hasMore:false,totalCount:5});
|
||||
});
|
||||
|
||||
it('keeps unread-first ordering, aggregates unread once, and isolates the other couple',async()=>{
|
||||
const page=await repo.matches(u1,0,5,false);
|
||||
// The unread match sorts ahead of newer ones; the rest stay newest-first.
|
||||
expect(page.matches.map(item=>item.id)).toEqual([paged[4],paged[0],paged[1],paged[2],paged[3]]);
|
||||
expect(page.matches[0]).toMatchObject({unreadCount:2});
|
||||
// Our own message is not unread to us, and everything else is zero.
|
||||
expect(page.matches.slice(1).every(item=>item.unreadCount===0)).toBe(true);
|
||||
// Unread is per viewer: the partner's own two messages are not unread to them,
|
||||
// but ours is, so the same thread reports a different count on each side.
|
||||
expect((await repo.matches(u2,0,5,false)).matches.find(item=>item.id===paged[4])?.unreadCount).toBe(1);
|
||||
// The other couple's match is invisible here, and their member sees only theirs.
|
||||
expect(page.matches.some(item=>item.id===otherMatch)).toBe(false);
|
||||
expect((await repo.matches(outsider,0,5,false)).matches.map(item=>item.id)).toEqual([otherMatch]);
|
||||
});
|
||||
|
||||
it('pages the archived view separately and per user',async()=>{
|
||||
await repo.archive(u1,paged[0],true);
|
||||
const active=await repo.matches(u1,0,5,false);
|
||||
expect(active.matches.some(item=>item.id===paged[0])).toBe(false);
|
||||
expect(active.totalCount).toBe(4);
|
||||
const archived=await repo.matches(u1,0,5,true);
|
||||
expect(archived.matches.map(item=>item.id)).toEqual([paged[0]]);
|
||||
expect(archived).toMatchObject({totalCount:1,hasMore:false});
|
||||
// Archiving is per user, so the partner still sees it in their active list.
|
||||
expect((await repo.matches(u2,0,5,false)).matches.some(item=>item.id===paged[0])).toBe(true);
|
||||
await repo.archive(u1,paged[0],false);
|
||||
});
|
||||
});
|
||||
|
||||
it('counts the day in the couple shared zone and prefers the first reported member zone',async()=>{await pool.query('update profiles set timezone=$2 where id=$1',[u1,'Pacific/Auckland']);await pool.query('update profiles set timezone=$2 where id=$1',[u2,'America/Los_Angeles']);
|
||||
await pool.query('delete from responses where couple_id=$1',[couple]);await pool.query('delete from matches where couple_id=$1',[couple]);
|
||||
await repo.submit(u1,{questionId:q,answer:'yes'});
|
||||
|
||||
@@ -10,7 +10,7 @@ const questionId='22222222-2222-4222-8222-222222222222';
|
||||
function repo():AnswersRepository{return {
|
||||
recommended:vi.fn(async()=>[]),pending:vi.fn(async()=>[]),answerGap:vi.fn(async()=>({unanswered_by_partner:0,threshold:10,is_blocked:false})),
|
||||
dailyLimit:vi.fn(async()=>({responses_today:0,limit_value:0,remaining:0,reset_at:null,is_blocked:false})),
|
||||
submit:vi.fn(async()=>({response:{} as never,match:null})),update:vi.fn(async()=>({success:true})),responses:vi.fn(async()=>({responses:[],totalCount:0})),matches:vi.fn(async()=>({matches:[],totalCount:0})),markSeen:vi.fn(async()=>undefined),archive:vi.fn(async()=>undefined),streak:vi.fn(async()=>null),close:vi.fn(async()=>undefined)};}
|
||||
submit:vi.fn(async()=>({response:{} as never,match:null})),update:vi.fn(async()=>({success:true})),responses:vi.fn(async()=>({responses:[],totalCount:0})),matches:vi.fn(async()=>({matches:[],totalCount:0,hasMore:false})),markSeen:vi.fn(async()=>undefined),archive:vi.fn(async()=>undefined),streak:vi.fn(async()=>null),close:vi.fn(async()=>undefined)};}
|
||||
function app(r:AnswersRepository){const app=new Hono<{Variables:{identity:AuthIdentity}}>();app.use('/v1/*',async(c,n)=>{c.set('identity',identity);await n();});registerAnswerRoutes(app,r);return app;}
|
||||
|
||||
describe('answer routes',()=>{
|
||||
@@ -18,6 +18,16 @@ describe('answer routes',()=>{
|
||||
it('maps confirmation and domain failures',async()=>{const r=repo();vi.mocked(r.update).mockResolvedValueOnce({success:false,requires_confirmation:true,message_count:2});const a=app(r);expect(await (await a.request(`/v1/responses/${questionId}`,{method:'PATCH',headers:{'content-type':'application/json'},body:'{"new_answer":"no"}'})).json()).toMatchObject({requires_confirmation:true,message_count:2});vi.mocked(r.submit).mockRejectedValueOnce(new AnswersError('daily_limit',429,'Daily response limit reached',{daily_limit:10,responses_today:10,remaining:0,reset_at:'2026-08-29T00:00:00.000Z'}));const blocked=await a.request('/v1/responses',{method:'POST',headers:{'content-type':'application/json'},body:JSON.stringify({question_id:questionId,answer:'yes'})});expect(blocked.status).toBe(429);
|
||||
// The client renders the limit screen straight from this body, so its shape is a contract.
|
||||
expect(await blocked.json()).toMatchObject({error:{code:'daily_limit'},details:{daily_limit:10,responses_today:10,remaining:0,reset_at:'2026-08-29T00:00:00.000Z'}});});
|
||||
it('reports pagination from the repository, including a page with no total',async()=>{const r=repo();const a=app(r);
|
||||
// Later pages skip the count entirely, so `hasMore` — not a client-side
|
||||
// comparison against totalCount — is what decides whether to keep paging.
|
||||
vi.mocked(r.matches).mockResolvedValueOnce({matches:[],totalCount:null,hasMore:true});
|
||||
const later=await a.request('/v1/matches?page=2&limit=20');
|
||||
expect(later.status).toBe(200);
|
||||
expect(await later.json()).toMatchObject({totalCount:null,hasMore:true});
|
||||
expect(r.matches).toHaveBeenCalledWith(identity.id,2,20,false);
|
||||
await a.request('/v1/matches?page=0&limit=20&archived=true');
|
||||
expect(r.matches).toHaveBeenLastCalledWith(identity.id,0,20,true);});
|
||||
it('scopes match mutations to identity',async()=>{const r=repo();const a=app(r);await a.request('/v1/matches/seen',{method:'PATCH',headers:{'content-type':'application/json'},body:JSON.stringify({ids:[questionId]})});expect(r.markSeen).toHaveBeenCalledWith(identity.id,[questionId]);await a.request(`/v1/matches/${questionId}/archive`,{method:'PUT',headers:{'content-type':'application/json'},body:'{"archived":true}'});expect(r.archive).toHaveBeenCalledWith(identity.id,questionId,true);});
|
||||
});
|
||||
|
||||
|
||||
@@ -75,6 +75,31 @@ describe.skipIf(!databaseUrl || !localDatabase)('API + PostgreSQL integration',
|
||||
},
|
||||
});
|
||||
|
||||
// Reading the profile again must not rewrite it. Every authenticated request the
|
||||
// app makes starts here, so an unconditional upsert turned a read into a write
|
||||
// on the busiest path in the API.
|
||||
const stamps = async () => (await pool.query<{ auth_synced_at: Date | null; updated_at: Date }>(
|
||||
'select auth_synced_at, updated_at from profiles where id = $1',
|
||||
['33333333-3333-4333-8333-333333333333'],
|
||||
)).rows[0];
|
||||
const created = await stamps();
|
||||
expect(created.auth_synced_at).not.toBeNull();
|
||||
|
||||
for (let attempt = 0; attempt < 3; attempt += 1) {
|
||||
expect((await app.request('/v1/me', { headers })).status).toBe(200);
|
||||
}
|
||||
const unchanged = await stamps();
|
||||
expect(unchanged.auth_synced_at?.getTime()).toBe(created.auth_synced_at?.getTime());
|
||||
expect(unchanged.updated_at.getTime()).toBe(created.updated_at.getTime());
|
||||
|
||||
// A genuine identity change still writes through, so first use and a changed
|
||||
// email are both recorded rather than skipped.
|
||||
await pool.query('update profiles set email = $2 where id = $1', ['33333333-3333-4333-8333-333333333333', 'stale@sauci.test']);
|
||||
expect((await app.request('/v1/me', { headers })).status).toBe(200);
|
||||
const resynced = await stamps();
|
||||
expect(resynced.auth_synced_at!.getTime()).toBeGreaterThan(created.auth_synced_at!.getTime());
|
||||
expect(await (await app.request('/v1/me', { headers })).json()).toMatchObject({ profile: { email: 'integration@sauci.test' } });
|
||||
|
||||
expect((await app.request('/v1/me/feature-interests/better-chat', { method: 'PUT', headers })).status).toBe(200);
|
||||
expect(await (await app.request('/v1/me/feature-interests/better-chat', { headers })).json())
|
||||
.toEqual({ feature: 'better-chat', interested: true });
|
||||
|
||||
@@ -30,6 +30,99 @@ describe.skipIf(!databaseUrl || !local)('app-data repository + PostgreSQL', () =
|
||||
await pool.query("insert into messages(id,match_id,user_id,media_path,media_type) values($1,$2,$3,'x.jpg','image')",[message,match,alice]);
|
||||
});
|
||||
afterAll(async()=>{await repo.close();await pool.end();await admin.query(`drop schema "${schema}" cascade`);await admin.end();});
|
||||
it('summarises change markers per couple without leaking another couple state',async()=>{
|
||||
// This one small read replaced a five-second refresh of profile, couple, matches,
|
||||
// pending questions, packs and unread counts, so its markers have to move for
|
||||
// exactly the changes those refreshes existed to notice.
|
||||
const before=await repo.syncSummary(alice);
|
||||
expect(before).toMatchObject({couple_id:couple,partner_id:bob,match_count:1,new_match_count:1,unread_total:0});
|
||||
expect(before.profile_updated_at).not.toBeNull();
|
||||
expect(Date.parse(before.server_time)).not.toBeNaN();
|
||||
|
||||
// The partner's own message is unread to them and not to us.
|
||||
expect((await repo.syncSummary(bob)).unread_total).toBe(1);
|
||||
|
||||
// A partner profile edit moves the partner marker, which is what tells the
|
||||
// client to re-read the couple rather than everything.
|
||||
await pool.query("update profiles set name='Bobby', updated_at=now() where id=$1",[bob]);
|
||||
const afterPartner=await repo.syncSummary(alice);
|
||||
expect(Date.parse(afterPartner.partner_updated_at!)).toBeGreaterThan(Date.parse(before.partner_updated_at!));
|
||||
expect(afterPartner.profile_updated_at).toBe(before.profile_updated_at);
|
||||
|
||||
// couple_packs carries no timestamp, so equal-and-opposite toggles have to be
|
||||
// caught by a digest rather than a max(updated_at).
|
||||
const emptyDigest=before.enabled_packs_fingerprint;
|
||||
expect(emptyDigest).toEqual(expect.any(String));
|
||||
await pool.query('insert into couple_packs(couple_id,pack_id,enabled) values($1,$2,true)',[couple,pack]);
|
||||
const enabled=await repo.syncSummary(alice);
|
||||
expect(enabled.enabled_packs_fingerprint).not.toBe(emptyDigest);
|
||||
await pool.query('update couple_packs set enabled=false where couple_id=$1 and pack_id=$2',[couple,pack]);
|
||||
expect((await repo.syncSummary(alice)).enabled_packs_fingerprint).toBe(emptyDigest);
|
||||
// Toggling back must land on the same digest, so the client stops refetching.
|
||||
await pool.query('update couple_packs set enabled=true where couple_id=$1 and pack_id=$2',[couple,pack]);
|
||||
expect((await repo.syncSummary(alice)).enabled_packs_fingerprint).toBe(enabled.enabled_packs_fingerprint);
|
||||
|
||||
// An outsider's summary is about their own couple and never ours.
|
||||
const other=await repo.syncSummary(outsider);
|
||||
expect(other).toMatchObject({couple_id:otherCouple,partner_id:null,match_count:0,unread_total:0});
|
||||
});
|
||||
|
||||
it('moves the match-state fingerprint on an in-place edit that the count and latest-at markers miss',async()=>{
|
||||
// Editing an existing match's type/response summary (response editing) creates
|
||||
// no new match and touches no created_at, so match_count/new_match_count/
|
||||
// latest_match_at all hold steady; only the state fingerprint must move.
|
||||
const before=await repo.syncSummary(alice);
|
||||
await pool.query("update matches set match_type='yes_maybe', response_summary=$2 where id=$1",[match,JSON.stringify({edited:true})]);
|
||||
const after=await repo.syncSummary(alice);
|
||||
expect(after.match_count).toBe(before.match_count);
|
||||
expect(after.new_match_count).toBe(before.new_match_count);
|
||||
expect(after.latest_match_at).toBe(before.latest_match_at);
|
||||
expect(after.match_state_fingerprint).not.toBe(before.match_state_fingerprint);
|
||||
// Reverting lands back on the same digest, so the client stops refetching.
|
||||
await pool.query("update matches set match_type='yes_yes', response_summary=null where id=$1",[match]);
|
||||
expect((await repo.syncSummary(alice)).match_state_fingerprint).toBe(before.match_state_fingerprint);
|
||||
});
|
||||
|
||||
it('moves the match-unread fingerprint when unread redistributes without changing the total',async()=>{
|
||||
const secondMatch='cccccccc-cccc-4ccc-8ccc-cccccccccccc';
|
||||
const secondQuestion='dddddddd-dddd-4ddd-8ddd-dddddddddddd';
|
||||
const originalUnread='eeeeeeee-eeee-4eee-8eee-eeeeeeeeeeee';
|
||||
const secondUnread='11111111-2222-4333-8444-555555555555';
|
||||
await pool.query("insert into questions(id,pack_id,text) values($1,$2,'Question 2')",[secondQuestion,pack]);
|
||||
await pool.query("insert into matches(id,couple_id,question_id,match_type) values($1,$2,$3,'yes_yes')",[secondMatch,couple,secondQuestion]);
|
||||
// Alice's own message ('message', from beforeAll) is never unread to her, so
|
||||
// give the original match a message from her partner to actually redistribute.
|
||||
await pool.query("insert into messages(id,match_id,user_id,media_path,media_type) values($1,$2,$3,'y.jpg','image')",[originalUnread,match,bob]);
|
||||
await pool.query("insert into messages(id,match_id,user_id,media_path,media_type) values($1,$2,$3,'y2.jpg','image')",[secondUnread,secondMatch,bob]);
|
||||
|
||||
const before=await repo.syncSummary(alice);
|
||||
expect(before.unread_total).toBe(2);
|
||||
|
||||
// Read the original match's unread message while a new one lands in the
|
||||
// second match: unread_total holds steady at 2, but which match owns it moves.
|
||||
await pool.query('update messages set read_at=now() where id=$1',[originalUnread]);
|
||||
const thirdMessage='ffffffff-ffff-4fff-8fff-ffffffffffff';
|
||||
await pool.query("insert into messages(id,match_id,user_id,media_path,media_type) values($1,$2,$3,'z.jpg','image')",[thirdMessage,secondMatch,bob]);
|
||||
|
||||
const after=await repo.syncSummary(alice);
|
||||
expect(after.unread_total).toBe(before.unread_total);
|
||||
expect(after.match_unread_fingerprint).not.toBe(before.match_unread_fingerprint);
|
||||
|
||||
// Clean up so later assertions in this suite see the original single-match state.
|
||||
await pool.query('delete from messages where id=any($1::uuid[])',[[originalUnread,secondUnread,thirdMessage]]);
|
||||
await pool.query('delete from matches where id=$1',[secondMatch]);
|
||||
await pool.query('delete from questions where id=$1',[secondQuestion]);
|
||||
});
|
||||
|
||||
it('moves the streak marker from couple_streaks.updated_at',async()=>{
|
||||
await pool.query('insert into couple_streaks(couple_id) values($1) on conflict do nothing',[couple]);
|
||||
const before=await repo.syncSummary(alice);
|
||||
expect(before.streak_updated_at).toEqual(expect.any(String));
|
||||
await pool.query('update couple_streaks set current_streak=current_streak+1, updated_at=now() where couple_id=$1',[couple]);
|
||||
const after=await repo.syncSummary(alice);
|
||||
expect(Date.parse(after.streak_updated_at!)).toBeGreaterThan(Date.parse(before.streak_updated_at!));
|
||||
});
|
||||
|
||||
it('returns only public catalog data and couple-owned match context',async()=>{
|
||||
expect((await repo.packQuestions(alice,pack))[0]?.text).toBe('Question');
|
||||
expect((await repo.matchContext(bob,match)).responses).toHaveLength(2);
|
||||
|
||||
@@ -16,6 +16,12 @@ function repository(): AppDataRepository {
|
||||
markMediaViewed: vi.fn(async () => ({ media_viewed_at: 'now', media_expires_at: null })),
|
||||
nudgeStatus: vi.fn(async () => ({ last_nudge_sent_at: null })),
|
||||
sendNudge: vi.fn(async () => ({ success: true as const, notification_sent: false, reason: 'no_push_token', next_nudge_available_at: 'later' })),
|
||||
syncSummary: vi.fn(async () => ({
|
||||
server_time: 'now', couple_id: null, profile_updated_at: null, partner_id: null, partner_updated_at: null,
|
||||
match_count: 0, new_match_count: 0, latest_match_at: null, pending_yours: 0, pending_theirs: 0,
|
||||
unread_total: 0, enabled_packs_fingerprint: '',
|
||||
match_state_fingerprint: '', match_unread_fingerprint: '', streak_updated_at: null,
|
||||
})),
|
||||
getLiveDraw: vi.fn(async () => ({ strokes: [], revision: 0, updated_at: null, updated_by: null })),
|
||||
putLiveDraw: vi.fn(async () => ({ strokes: [], revision: 1, updated_at: 'now', updated_by: identity.id })),
|
||||
close: vi.fn(async () => undefined),
|
||||
|
||||
@@ -48,7 +48,7 @@ describe.skipIf(!databaseUrl || !isLocal)('chat repository + PostgreSQL', () =>
|
||||
|
||||
it('sends, lists, counts and reads text messages for couple members only', async () => {
|
||||
const sent = await repository.sendText(alice, match, 'Hello');
|
||||
expect((await repository.listMessages(bob, match))[0]?.content).toBe('Hello');
|
||||
expect((await repository.listMessages(bob, match)).messages[0]?.content).toBe('Hello');
|
||||
expect(await repository.unreadCounts(bob)).toEqual({ total: 1, by_match: { [match]: 1 } });
|
||||
expect((await repository.markDelivered(bob, sent.id)).delivered_at).not.toBeNull();
|
||||
expect((await repository.markMatchRead(bob, match)).updated).toBe(1);
|
||||
@@ -58,14 +58,110 @@ describe.skipIf(!databaseUrl || !isLocal)('chat repository + PostgreSQL', () =>
|
||||
it('enforces per-user and author-only deletions plus reports', async () => {
|
||||
const sent = await repository.sendText(alice, match, 'Moderate me');
|
||||
await repository.deleteForSelf(bob, sent.id);
|
||||
expect((await repository.listMessages(bob, match)).some(item => item.id === sent.id)).toBe(false);
|
||||
expect((await repository.listMessages(alice, match)).some(item => item.id === sent.id)).toBe(true);
|
||||
expect((await repository.listMessages(bob, match)).messages.some(item => item.id === sent.id)).toBe(false);
|
||||
expect((await repository.listMessages(alice, match)).messages.some(item => item.id === sent.id)).toBe(true);
|
||||
await expect(repository.deleteForEveryone(bob, sent.id)).rejects.toMatchObject({ code: 'not_message_author' });
|
||||
expect((await repository.deleteForEveryone(alice, sent.id)).deleted_at).not.toBeNull();
|
||||
await repository.report(bob, sent.id, 'harassment');
|
||||
await expect(repository.report(bob, sent.id, 'spam')).rejects.toBeInstanceOf(ChatError);
|
||||
});
|
||||
|
||||
it('answers a delta with only what changed, and the same rows a full read would give', async () => {
|
||||
const before = await repository.listMessages(bob, match);
|
||||
const cursor = before.server_time;
|
||||
expect(before.complete).toBe(true);
|
||||
|
||||
// Nothing has happened since the cursor, so the delta carries no row the client
|
||||
// has not already seen. The cursor is deliberately held a couple of seconds
|
||||
// behind wall clock so a write racing the read is repeated rather than lost;
|
||||
// that repetition is absorbed by merging on id, never by blanking the thread.
|
||||
const known = new Set(before.messages.map(item => item.id));
|
||||
const quiet = await repository.listMessages(bob, match, { since: cursor });
|
||||
expect(quiet.complete).toBe(false);
|
||||
expect(quiet.messages.every(item => known.has(item.id))).toBe(true);
|
||||
expect(quiet.removed_ids?.every(id => !known.has(id))).toBe(true);
|
||||
|
||||
const fresh = await repository.sendText(alice, match, 'Delta arrival');
|
||||
const delta = await repository.listMessages(bob, match, { since: cursor });
|
||||
expect(delta.messages.map(item => item.id)).toContain(fresh.id);
|
||||
// Media and moderation columns must survive the narrower query.
|
||||
expect(delta.messages.find(item => item.id === fresh.id)).toMatchObject({
|
||||
content: 'Delta arrival', media_path: null, media_type: null, media_expired: false, read_at: null,
|
||||
});
|
||||
|
||||
// A status change with no new row still reaches the sender through the delta.
|
||||
const afterSend = await repository.listMessages(alice, match);
|
||||
await repository.markMatchRead(bob, match);
|
||||
const readDelta = await repository.listMessages(alice, match, { since: afterSend.server_time });
|
||||
expect(readDelta.messages.find(item => item.id === fresh.id)?.read_at).not.toBeNull();
|
||||
|
||||
// So does a deletion, which arrives as a removal rather than a silent gap.
|
||||
const doomed = await repository.sendText(alice, match, 'Delete me');
|
||||
const beforeDelete = await repository.listMessages(bob, match);
|
||||
await repository.deleteForSelf(bob, doomed.id);
|
||||
const removal = await repository.listMessages(bob, match, { since: beforeDelete.server_time });
|
||||
expect(removal.removed_ids).toContain(doomed.id);
|
||||
|
||||
// Ordering is the contract the screen renders against: newest first, both ways.
|
||||
const full = await repository.listMessages(bob, match);
|
||||
const timestamps = full.messages.map(item => Date.parse(item.created_at));
|
||||
expect([...timestamps].sort((a, b) => b - a)).toEqual(timestamps);
|
||||
|
||||
// Access control is unchanged on the delta path.
|
||||
await expect(repository.listMessages(outsider, match, { since: cursor })).rejects.toMatchObject({ code: 'match_not_found' });
|
||||
});
|
||||
|
||||
it('surfaces an expiring video to a chat that is already open, and only in the window it expires', async () => {
|
||||
// Expiry is the one visible change no write announces: the deadline simply
|
||||
// passes. Backdated so neither `created_at` nor `media_viewed_at` can be what
|
||||
// puts the row in a delta — only the deadline crossing may.
|
||||
const video = await repository.sendText(alice, match, 'Sent a video');
|
||||
await pool.query(
|
||||
`update messages
|
||||
set media_path = 'media:expiring', media_type = 'video',
|
||||
moderation_status = 'safe', encrypted_content = 'cipher', encryption_iv = 'iv',
|
||||
created_at = now() - interval '1 hour',
|
||||
media_viewed_at = now() - interval '1 hour',
|
||||
media_expires_at = now() + interval '1 hour'
|
||||
where id = $1`,
|
||||
[video.id],
|
||||
);
|
||||
|
||||
const before = await repository.listMessages(bob, match);
|
||||
expect(before.messages.find(item => item.id === video.id)).toMatchObject({ media_expired: false });
|
||||
|
||||
// A deadline still in the future is not a change, so the row is not re-sent on
|
||||
// every poll until it arrives.
|
||||
const quiet = await repository.listMessages(bob, match, { since: before.server_time });
|
||||
expect(quiet.messages.some(item => item.id === video.id)).toBe(false);
|
||||
|
||||
// The deadline now falls inside this poll's window.
|
||||
await pool.query(`update messages set media_expires_at = now() - interval '1 second' where id = $1`, [video.id]);
|
||||
const delta = await repository.listMessages(bob, match, { since: before.server_time });
|
||||
const expired = delta.messages.find(item => item.id === video.id);
|
||||
// Media, encryption and moderation columns still arrive whole on the delta path.
|
||||
expect(expired).toMatchObject({
|
||||
media_expired: true, media_type: 'video', media_path: 'media:expiring',
|
||||
moderation_status: 'safe', encrypted_content: 'cipher', encryption_iv: 'iv',
|
||||
});
|
||||
expect(expired?.media_expires_at).not.toBeNull();
|
||||
expect(expired?.media_viewed_at).not.toBeNull();
|
||||
|
||||
// A full read agrees, so leaving and re-entering the chat shows the same thing.
|
||||
expect((await repository.listMessages(bob, match)).messages.find(item => item.id === video.id))
|
||||
.toMatchObject({ media_expired: true });
|
||||
});
|
||||
|
||||
it('folds the partner typing check into the message read when asked', async () => {
|
||||
await repository.setTyping(alice, match, 4_000);
|
||||
const withTyping = await repository.listMessages(bob, match, { includeTyping: true });
|
||||
expect(withTyping.typing?.typing).toBe(true);
|
||||
// Own typing is never reflected back, so the sender does not see themselves.
|
||||
expect((await repository.listMessages(alice, match, { includeTyping: true })).typing?.typing).toBe(false);
|
||||
// And it stays absent unless requested, so the default read costs no extra work.
|
||||
expect((await repository.listMessages(bob, match)).typing).toBeUndefined();
|
||||
});
|
||||
|
||||
it('stores typing state with a short expiry and exposes only the partner state', async () => {
|
||||
await repository.setTyping(alice, match, 4_000);
|
||||
expect((await repository.getPartnerTyping(bob, match)).typing).toBe(true);
|
||||
|
||||
@@ -11,7 +11,8 @@ const message = { id: messageId, match_id: matchId, user_id: identity.id, conten
|
||||
|
||||
function setup() {
|
||||
const repository: ChatRepository = {
|
||||
listMessages: vi.fn(async () => [message]), sendText: vi.fn(async () => message),
|
||||
listMessages: vi.fn(async () => ({ messages: [message], removed_ids: [], server_time: new Date().toISOString(), complete: true })),
|
||||
sendText: vi.fn(async () => message),
|
||||
unreadCounts: vi.fn(async () => ({ total: 1, by_match: { [matchId]: 1 } })),
|
||||
markMatchRead: vi.fn(async () => ({ updated: 1, read_at: new Date().toISOString() })),
|
||||
markDelivered: vi.fn(async () => message), deleteForSelf: vi.fn(async () => undefined),
|
||||
@@ -43,6 +44,28 @@ describe('chat routes', () => {
|
||||
expect(repository.sendText).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it('passes a delta cursor and folded typing check through to the repository', async () => {
|
||||
const { app, repository } = setup();
|
||||
const since = '2026-08-28T10:00:00.000Z';
|
||||
const delta = await app.request(`/v1/matches/${matchId}/messages?since=${encodeURIComponent(since)}&typing=true`);
|
||||
expect(delta.status).toBe(200);
|
||||
expect(repository.listMessages).toHaveBeenCalledWith(identity.id, matchId, { since, includeTyping: true });
|
||||
|
||||
// Omitting both keeps the full-snapshot contract the client had before.
|
||||
const snapshot = await app.request(`/v1/matches/${matchId}/messages`);
|
||||
expect(snapshot.status).toBe(200);
|
||||
expect(await snapshot.json()).toMatchObject({ messages: [{ id: messageId }] });
|
||||
expect(repository.listMessages).toHaveBeenLastCalledWith(identity.id, matchId, { includeTyping: false });
|
||||
});
|
||||
|
||||
it('refuses a malformed cursor rather than silently reading everything', async () => {
|
||||
const { app, repository } = setup();
|
||||
const response = await app.request(`/v1/matches/${matchId}/messages?since=yesterday`);
|
||||
expect(response.status).toBe(400);
|
||||
expect(await response.json()).toMatchObject({ error: { code: 'invalid_since' } });
|
||||
expect(repository.listMessages).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it('supports unread, typing, deletion and reporting actions', async () => {
|
||||
const { app, repository } = setup();
|
||||
expect((await app.request('/v1/chat/unread')).status).toBe(200);
|
||||
|
||||
@@ -0,0 +1,49 @@
|
||||
import { Pool } from 'pg';
|
||||
import { afterEach, describe, expect, it, vi } from 'vitest';
|
||||
import {
|
||||
DEFAULT_POOL_MAX,
|
||||
closeResolvedPool,
|
||||
createDatabasePool,
|
||||
resolvePool,
|
||||
} from '../src/db/pool.js';
|
||||
|
||||
const url = 'postgres://user:pass@127.0.0.1:5432/sauci_test';
|
||||
|
||||
afterEach(() => vi.restoreAllMocks());
|
||||
|
||||
describe('database pool', () => {
|
||||
it('bounds the shared pool and only forwards timeouts that were configured', () => {
|
||||
// A process-wide pool is the whole point of the change: an unbounded default
|
||||
// multiplied by one pool per repository is what the self-hosted database had
|
||||
// to absorb before.
|
||||
const bounded = createDatabasePool(url, { max: 4, idleTimeoutMillis: 1_000, connectionTimeoutMillis: 2_000 });
|
||||
expect(bounded.options.max).toBe(4);
|
||||
expect(bounded.options.idleTimeoutMillis).toBe(1_000);
|
||||
expect(bounded.options.connectionTimeoutMillis).toBe(2_000);
|
||||
|
||||
const defaults = createDatabasePool(url);
|
||||
expect(defaults.options.max).toBe(DEFAULT_POOL_MAX);
|
||||
|
||||
return Promise.all([bounded.end(), defaults.end()]);
|
||||
});
|
||||
|
||||
it('owns a pool it built from a URL and never ends one it was handed', async () => {
|
||||
// Repositories in the API process share one pool, so closing any single
|
||||
// repository must not take the connections out from under the others. Tests
|
||||
// still construct from a URL and must be able to close what they created.
|
||||
const owned = resolvePool(url);
|
||||
expect(owned.owned).toBe(true);
|
||||
const ownedEnd = vi.spyOn(owned.pool, 'end');
|
||||
await closeResolvedPool(owned.pool, owned.owned);
|
||||
expect(ownedEnd).toHaveBeenCalledTimes(1);
|
||||
|
||||
const shared = new Pool({ connectionString: url });
|
||||
const borrowed = resolvePool(shared);
|
||||
expect(borrowed.owned).toBe(false);
|
||||
expect(borrowed.pool).toBe(shared);
|
||||
const sharedEnd = vi.spyOn(shared, 'end');
|
||||
await closeResolvedPool(borrowed.pool, borrowed.owned);
|
||||
expect(sharedEnd).not.toHaveBeenCalled();
|
||||
await shared.end();
|
||||
});
|
||||
});
|
||||
Reference in new issue
Block a user