feat: audit cleanup (#21567)

server/src/dtos/sync.dto.ts

@@ -336,6 +336,9 @@ export class SyncAckV1 {}
 @ExtraModel()
 export class SyncResetV1 {}
 
+@ExtraModel()
+export class SyncCompleteV1 {}
+
 export type SyncItem = {
   [SyncEntityType.AuthUserV1]: SyncAuthUserV1;
   [SyncEntityType.UserV1]: SyncUserV1;
@@ -382,6 +385,7 @@ export type SyncItem = {
   [SyncEntityType.UserMetadataV1]: SyncUserMetadataV1;
   [SyncEntityType.UserMetadataDeleteV1]: SyncUserMetadataDeleteV1;
   [SyncEntityType.SyncAckV1]: SyncAckV1;
+  [SyncEntityType.SyncCompleteV1]: SyncCompleteV1;
   [SyncEntityType.SyncResetV1]: SyncResetV1;
 };
 

server/src/enum.ts

@@ -530,6 +530,7 @@ export enum JobName {
   AssetGenerateThumbnails = 'AssetGenerateThumbnails',
 
   AuditLogCleanup = 'AuditLogCleanup',
+  AuditTableCleanup = 'AuditTableCleanup',
 
   DatabaseBackup = 'DatabaseBackup',
 
@@ -708,6 +709,7 @@ export enum SyncEntityType {
 
   SyncAckV1 = 'SyncAckV1',
   SyncResetV1 = 'SyncResetV1',
+  SyncCompleteV1 = 'SyncCompleteV1',
 }
 
 export enum NotificationLevel {

server/src/queries/sync.repository.sql

@@ -957,7 +957,7 @@ where
 order by
   "stack"."updateId" asc
 
--- SyncRepository.people.getDeletes
+-- SyncRepository.person.getDeletes
 select
   "id",
   "personId"
@@ -970,7 +970,7 @@ where
 order by
   "person_audit"."id" asc
 
--- SyncRepository.people.getUpserts
+-- SyncRepository.person.getUpserts
 select
   "id",
   "createdAt",

server/src/repositories/sync.repository.ts

@@ -1,5 +1,5 @@
 import { Injectable } from '@nestjs/common';
-import { Kysely } from 'kysely';
+import { Kysely, sql } from 'kysely';
 import { InjectKysely } from 'nestjs-kysely';
 import { columns } from 'src/database';
 import { DummyValue, GenerateSql } from 'src/decorators';
@@ -62,7 +62,7 @@ export class SyncRepository {
   partnerAsset: PartnerAssetsSync;
   partnerAssetExif: PartnerAssetExifsSync;
   partnerStack: PartnerStackSync;
-  people: PersonSync;
+  person: PersonSync;
   stack: StackSync;
   user: UserSync;
   userMetadata: UserMetadataSync;
@@ -84,7 +84,7 @@ export class SyncRepository {
     this.partnerAsset = new PartnerAssetsSync(this.db);
     this.partnerAssetExif = new PartnerAssetExifsSync(this.db);
     this.partnerStack = new PartnerStackSync(this.db);
-    this.people = new PersonSync(this.db);
+    this.person = new PersonSync(this.db);
     this.stack = new StackSync(this.db);
     this.user = new UserSync(this.db);
     this.userMetadata = new UserMetadataSync(this.db);
@@ -117,6 +117,15 @@ class BaseSync {
       .orderBy(idRef, 'asc');
   }
 
+  protected auditCleanup<T extends keyof DB>(t: T, days: number) {
+    const { table, ref } = this.db.dynamic;
+
+    return this.db
+      .deleteFrom(table(t).as(t))
+      .where(ref(`${t}.deletedAt`), '<', sql.raw(`now() - interval '${days} days'`))
+      .execute();
+  }
+
   protected upsertQuery<T extends keyof DB>(t: T, { nowId, ack }: SyncQueryOptions) {
     const { table, ref } = this.db.dynamic;
     const updateIdRef = ref(`${t}.updateId`);
@@ -150,6 +159,10 @@ class AlbumSync extends BaseSync {
       .stream();
   }
 
+  cleanupAuditTable(daysAgo: number) {
+    return this.auditCleanup('album_audit', daysAgo);
+  }
+
   @GenerateSql({ params: [dummyQueryOptions], stream: true })
   getUpserts(options: SyncQueryOptions) {
     const userId = options.userId;
@@ -286,6 +299,10 @@ class AlbumToAssetSync extends BaseSync {
       .stream();
   }
 
+  cleanupAuditTable(daysAgo: number) {
+    return this.auditCleanup('album_asset_audit', daysAgo);
+  }
+
   @GenerateSql({ params: [dummyQueryOptions], stream: true })
   getUpserts(options: SyncQueryOptions) {
     const userId = options.userId;
@@ -334,6 +351,10 @@ class AlbumUserSync extends BaseSync {
       .stream();
   }
 
+  cleanupAuditTable(daysAgo: number) {
+    return this.auditCleanup('album_user_audit', daysAgo);
+  }
+
   @GenerateSql({ params: [dummyQueryOptions], stream: true })
   getUpserts(options: SyncQueryOptions) {
     const userId = options.userId;
@@ -371,6 +392,10 @@ class AssetSync extends BaseSync {
       .stream();
   }
 
+  cleanupAuditTable(daysAgo: number) {
+    return this.auditCleanup('asset_audit', daysAgo);
+  }
+
   @GenerateSql({ params: [dummyQueryOptions], stream: true })
   getUpserts(options: SyncQueryOptions) {
     return this.upsertQuery('asset', options)
@@ -400,6 +425,10 @@ class PersonSync extends BaseSync {
       .stream();
   }
 
+  cleanupAuditTable(daysAgo: number) {
+    return this.auditCleanup('person_audit', daysAgo);
+  }
+
   @GenerateSql({ params: [dummyQueryOptions], stream: true })
   getUpserts(options: SyncQueryOptions) {
     return this.upsertQuery('person', options)
@@ -431,6 +460,10 @@ class AssetFaceSync extends BaseSync {
       .stream();
   }
 
+  cleanupAuditTable(daysAgo: number) {
+    return this.auditCleanup('asset_face_audit', daysAgo);
+  }
+
   @GenerateSql({ params: [dummyQueryOptions], stream: true })
   getUpserts(options: SyncQueryOptions) {
     return this.upsertQuery('asset_face', options)
@@ -473,6 +506,10 @@ class MemorySync extends BaseSync {
       .stream();
   }
 
+  cleanupAuditTable(daysAgo: number) {
+    return this.auditCleanup('memory_audit', daysAgo);
+  }
+
   @GenerateSql({ params: [dummyQueryOptions], stream: true })
   getUpserts(options: SyncQueryOptions) {
     return this.upsertQuery('memory', options)
@@ -505,6 +542,10 @@ class MemoryToAssetSync extends BaseSync {
       .stream();
   }
 
+  cleanupAuditTable(daysAgo: number) {
+    return this.auditCleanup('memory_asset_audit', daysAgo);
+  }
+
   @GenerateSql({ params: [dummyQueryOptions], stream: true })
   getUpserts(options: SyncQueryOptions) {
     return this.upsertQuery('memory_asset', options)
@@ -537,6 +578,10 @@ class PartnerSync extends BaseSync {
       .stream();
   }
 
+  cleanupAuditTable(daysAgo: number) {
+    return this.auditCleanup('partner_audit', daysAgo);
+  }
+
   @GenerateSql({ params: [dummyQueryOptions], stream: true })
   getUpserts(options: SyncQueryOptions) {
     const userId = options.userId;
@@ -616,6 +661,10 @@ class StackSync extends BaseSync {
       .stream();
   }
 
+  cleanupAuditTable(daysAgo: number) {
+    return this.auditCleanup('stack_audit', daysAgo);
+  }
+
   @GenerateSql({ params: [dummyQueryOptions], stream: true })
   getUpserts(options: SyncQueryOptions) {
     return this.upsertQuery('stack', options)
@@ -664,6 +713,10 @@ class UserSync extends BaseSync {
     return this.auditQuery('user_audit', options).select(['id', 'userId']).stream();
   }
 
+  cleanupAuditTable(daysAgo: number) {
+    return this.auditCleanup('user_audit', daysAgo);
+  }
+
   @GenerateSql({ params: [dummyQueryOptions], stream: true })
   getUpserts(options: SyncQueryOptions) {
     return this.upsertQuery('user', options).select(columns.syncUser).stream();
@@ -679,6 +732,10 @@ class UserMetadataSync extends BaseSync {
       .stream();
   }
 
+  cleanupAuditTable(daysAgo: number) {
+    return this.auditCleanup('user_metadata_audit', daysAgo);
+  }
+
   @GenerateSql({ params: [dummyQueryOptions], stream: true })
   getUpserts(options: SyncQueryOptions) {
     return this.upsertQuery('user_metadata', options)
@@ -698,6 +755,10 @@ class AssetMetadataSync extends BaseSync {
       .stream();
   }
 
+  cleanupAuditTable(daysAgo: number) {
+    return this.auditCleanup('asset_metadata_audit', daysAgo);
+  }
+
   @GenerateSql({ params: [dummyQueryOptions, DummyValue.UUID], stream: true })
   getUpserts(options: SyncQueryOptions, userId: string) {
     return this.upsertQuery('asset_metadata', options)

server/src/schema/index.ts

@@ -166,6 +166,7 @@ export interface DB {
   api_key: ApiKeyTable;
 
   asset: AssetTable;
+  asset_audit: AssetAuditTable;
   asset_exif: AssetExifTable;
   asset_face: AssetFaceTable;
   asset_face_audit: AssetFaceAuditTable;
@@ -173,7 +174,6 @@ export interface DB {
   asset_metadata: AssetMetadataTable;
   asset_metadata_audit: AssetMetadataAuditTable;
   asset_job_status: AssetJobStatusTable;
-  asset_audit: AssetAuditTable;
 
   audit: AuditTable;
 

server/src/schema/tables/memory-asset-audit.table.ts

@@ -1,11 +1,11 @@
 import { PrimaryGeneratedUuidV7Column } from 'src/decorators';
 import { MemoryTable } from 'src/schema/tables/memory.table';
-import { Column, CreateDateColumn, ForeignKeyColumn, Table } from 'src/sql-tools';
+import { Column, CreateDateColumn, ForeignKeyColumn, Generated, Table, Timestamp } from 'src/sql-tools';
 
 @Table('memory_asset_audit')
 export class MemoryAssetAuditTable {
   @PrimaryGeneratedUuidV7Column()
-  id!: string;
+  id!: Generated<string>;
 
   @ForeignKeyColumn(() => MemoryTable, { type: 'uuid', onDelete: 'CASCADE', onUpdate: 'CASCADE' })
   memoryId!: string;
@@ -14,5 +14,5 @@ export class MemoryAssetAuditTable {
   assetId!: string;
 
   @CreateDateColumn({ default: () => 'clock_timestamp()', index: true })
-  deletedAt!: Date;
+  deletedAt!: Generated<Timestamp>;
 }

server/src/services/job.service.spec.ts

@@ -42,6 +42,7 @@ describe(JobService.name, () => {
         { name: JobName.PersonCleanup },
         { name: JobName.MemoryCleanup },
         { name: JobName.SessionCleanup },
+        { name: JobName.AuditTableCleanup },
         { name: JobName.AuditLogCleanup },
         { name: JobName.MemoryGenerate },
         { name: JobName.UserSyncUsage },

server/src/services/job.service.ts

@@ -281,6 +281,7 @@ export class JobService extends BaseService {
         { name: JobName.PersonCleanup },
         { name: JobName.MemoryCleanup },
         { name: JobName.SessionCleanup },
+        { name: JobName.AuditTableCleanup },
         { name: JobName.AuditLogCleanup },
       );
     }

server/src/services/sync.service.ts

@@ -1,8 +1,9 @@
 import { BadRequestException, ForbiddenException, Injectable } from '@nestjs/common';
 import { Insertable } from 'kysely';
-import { DateTime } from 'luxon';
+import { DateTime, Duration } from 'luxon';
 import { Writable } from 'node:stream';
 import { AUDIT_LOG_MAX_DURATION } from 'src/constants';
+import { OnJob } from 'src/decorators';
 import { AssetResponseDto, mapAsset } from 'src/dtos/asset-response.dto';
 import { AuthDto } from 'src/dtos/auth.dto';
 import {
@@ -15,7 +16,16 @@ import {
   SyncItem,
   SyncStreamDto,
 } from 'src/dtos/sync.dto';
-import { AssetVisibility, DatabaseAction, EntityType, Permission, SyncEntityType, SyncRequestType } from 'src/enum';
+import {
+  AssetVisibility,
+  DatabaseAction,
+  EntityType,
+  JobName,
+  Permission,
+  QueueName,
+  SyncEntityType,
+  SyncRequestType,
+} from 'src/enum';
 import { SyncQueryOptions } from 'src/repositories/sync.repository';
 import { SessionSyncCheckpointTable } from 'src/schema/tables/sync-checkpoint.table';
 import { BaseService } from 'src/services/base.service';
@@ -32,6 +42,8 @@ type AssetLike = Omit<SyncAssetV1, 'checksum' | 'thumbhash'> & {
 };
 
 const COMPLETE_ID = 'complete';
+const MAX_DAYS = 30;
+const MAX_DURATION = Duration.fromObject({ days: MAX_DAYS });
 
 const mapSyncAssetV1 = ({ checksum, thumbhash, ...data }: AssetLike): SyncAssetV1 => ({
   ...data,
@@ -137,19 +149,24 @@ export class SyncService extends BaseService {
     }
 
     const isPendingSyncReset = await this.sessionRepository.isPendingSyncReset(session.id);
-
     if (isPendingSyncReset) {
       send(response, { type: SyncEntityType.SyncResetV1, ids: ['reset'], data: {} });
       response.end();
       return;
     }
 
+    const checkpoints = await this.syncCheckpointRepository.getAll(session.id);
+    const checkpointMap: CheckpointMap = Object.fromEntries(checkpoints.map(({ type, ack }) => [type, fromAck(ack)]));
+
+    if (this.needsFullSync(checkpointMap)) {
+      send(response, { type: SyncEntityType.SyncResetV1, ids: ['reset'], data: {} });
+      response.end();
+      return;
+    }
+
     const { nowId } = await this.syncCheckpointRepository.getNow();
     const options: SyncQueryOptions = { nowId, userId: auth.user.id };
 
-    const checkpoints = await this.syncCheckpointRepository.getAll(session.id);
-    const checkpointMap: CheckpointMap = Object.fromEntries(checkpoints.map(({ type, ack }) => [type, fromAck(ack)]));
-
     const handlers: Record<SyncRequestType, () => Promise<void>> = {
       [SyncRequestType.AuthUsersV1]: () => this.syncAuthUsersV1(options, response, checkpointMap),
       [SyncRequestType.UsersV1]: () => this.syncUsersV1(options, response, checkpointMap),
@@ -180,9 +197,41 @@ export class SyncService extends BaseService {
       await handler();
     }
 
+    send(response, { type: SyncEntityType.SyncCompleteV1, ids: [nowId], data: {} });
+
     response.end();
   }
 
+  @OnJob({ name: JobName.AuditTableCleanup, queue: QueueName.BackgroundTask })
+  async onAuditTableCleanup() {
+    const pruneThreshold = MAX_DAYS + 1;
+
+    await this.syncRepository.album.cleanupAuditTable(pruneThreshold);
+    await this.syncRepository.albumUser.cleanupAuditTable(pruneThreshold);
+    await this.syncRepository.albumToAsset.cleanupAuditTable(pruneThreshold);
+    await this.syncRepository.asset.cleanupAuditTable(pruneThreshold);
+    await this.syncRepository.assetFace.cleanupAuditTable(pruneThreshold);
+    await this.syncRepository.assetMetadata.cleanupAuditTable(pruneThreshold);
+    await this.syncRepository.memory.cleanupAuditTable(pruneThreshold);
+    await this.syncRepository.memoryToAsset.cleanupAuditTable(pruneThreshold);
+    await this.syncRepository.partner.cleanupAuditTable(pruneThreshold);
+    await this.syncRepository.person.cleanupAuditTable(pruneThreshold);
+    await this.syncRepository.stack.cleanupAuditTable(pruneThreshold);
+    await this.syncRepository.user.cleanupAuditTable(pruneThreshold);
+    await this.syncRepository.userMetadata.cleanupAuditTable(pruneThreshold);
+  }
+
+  private needsFullSync(checkpointMap: CheckpointMap) {
+    const completeAck = checkpointMap[SyncEntityType.SyncCompleteV1];
+    if (!completeAck) {
+      return false;
+    }
+
+    const milliseconds = Number.parseInt(completeAck.updateId.replaceAll('-', '').slice(0, 12), 16);
+
+    return DateTime.fromMillis(milliseconds) < DateTime.now().minus(MAX_DURATION);
+  }
+
   private async syncAuthUsersV1(options: SyncQueryOptions, response: Writable, checkpointMap: CheckpointMap) {
     const upsertType = SyncEntityType.AuthUserV1;
     const upserts = this.syncRepository.authUser.getUpserts({ ...options, ack: checkpointMap[upsertType] });
@@ -719,13 +768,13 @@ export class SyncService extends BaseService {
 
   private async syncPeopleV1(options: SyncQueryOptions, response: Writable, checkpointMap: CheckpointMap) {
     const deleteType = SyncEntityType.PersonDeleteV1;
-    const deletes = this.syncRepository.people.getDeletes({ ...options, ack: checkpointMap[deleteType] });
+    const deletes = this.syncRepository.person.getDeletes({ ...options, ack: checkpointMap[deleteType] });
     for await (const { id, ...data } of deletes) {
       send(response, { type: deleteType, ids: [id], data });
     }
 
     const upsertType = SyncEntityType.PersonV1;
-    const upserts = this.syncRepository.people.getUpserts({ ...options, ack: checkpointMap[upsertType] });
+    const upserts = this.syncRepository.person.getUpserts({ ...options, ack: checkpointMap[upsertType] });
     for await (const { updateId, ...data } of upserts) {
       send(response, { type: upsertType, ids: [updateId], data });
     }

server/src/types.ts

@@ -275,6 +275,9 @@ export interface QueueStatus {
 }
 
 export type JobItem =
+  // Audit
+  | { name: JobName.AuditTableCleanup; data?: IBaseJob }
+
   // Backups
   | { name: JobName.DatabaseBackup; data?: IBaseJob }