feat: added archival management to AttachmentQueue via AttachmentContext

Author: Amine

packages/attachments/src/AttachmentContext.ts

@@ -17,17 +17,21 @@ export class AttachmentContext {
   /** Logger instance for diagnostic information */
   logger: ILogger;
 
+  /** Maximum number of archived attachments to keep before cleanup */
+  archivedCacheLimit: number = 100;
+
   /**
    * Creates a new AttachmentContext instance.
    * 
    * @param db - PowerSync database instance
    * @param tableName - Name of the table storing attachment records. Default: 'attachments'
    * @param logger - Logger instance for diagnostic output
    */
-  constructor(db: AbstractPowerSyncDatabase, tableName: string = 'attachments', logger: ILogger) {
+  constructor(db: AbstractPowerSyncDatabase, tableName: string = 'attachments', logger: ILogger, archivedCacheLimit: number) {
     this.db = db;
     this.tableName = tableName;
     this.logger = logger;
+    this.archivedCacheLimit = archivedCacheLimit;
   }
 
   /**
@@ -191,6 +195,53 @@ export class AttachmentContext {
     );
   }
 
+  async clearQueue(): Promise<void> {
+    await this.db.writeTransaction((tx) =>
+      tx.execute(
+        /* sql */
+        `
+          DELETE FROM ${this.tableName}
+        `
+      )
+    );
+  }
+
+  async deleteArchivedAttachments(callback?: (attachments: AttachmentRecord[]) => Promise<void>): Promise<boolean> {
+    const limit = 1000;
+
+    const results = await this.db.getAll(
+      /* sql */
+      `
+        SELECT * FROM ${this.tableName} WHERE state = ? ORDER BY timestamp DESC LIMIT ? OFFSET ?
+      `,
+      [
+        AttachmentState.ARCHIVED,
+        limit,
+        this.archivedCacheLimit,
+      ],
+    );
+
+    const archivedAttachments = results.map(attachmentFromSql);
+    if (archivedAttachments.length === 0) return false;
+
+    this.logger.info(`Deleting ${archivedAttachments.length} archived attachments. Archived attachment exceeds cache archiveCacheLimit of ${this.archivedCacheLimit}.`);
+
+    await callback?.(archivedAttachments);
+
+    const ids = archivedAttachments.map(attachment => attachment.id);
+
+    await this.db.executeBatch(
+      /* sql */
+      `
+        DELETE FROM ${this.tableName} WHERE id IN (?)
+      `,
+      [ids]
+    );
+
+    this.logger.info(`Deleted ${archivedAttachments.length} archived attachments`);
+    return archivedAttachments.length < limit;
+  }
+
   /**
    * Saves multiple attachment records in a single transaction.
    * 

packages/attachments/src/AttachmentQueue.ts

@@ -97,18 +97,18 @@ export class AttachmentQueue {
     archivedCacheLimit?: number;
     errorHandler?: AttachmentErrorHandler;
   }) {
-    this.context = new AttachmentContext(db, tableName, logger ?? db.logger);
     this.remoteStorage = remoteStorage;
     this.localStorage = localStorage;
     this.watchAttachments = watchAttachments;
     this.tableName = tableName;
-    this.syncingService = new SyncingService(this.context, localStorage, remoteStorage, logger ?? db.logger, errorHandler);
-    this.attachmentService = new AttachmentService(tableName, db);
     this.syncIntervalMs = syncIntervalMs;
     this.syncThrottleDuration = syncThrottleDuration;
-    this.watchActiveAttachments = this.attachmentService.watchActiveAttachments({ throttleMs: this.syncThrottleDuration });
-    this.downloadAttachments = downloadAttachments;
     this.archivedCacheLimit = archivedCacheLimit;
+    this.downloadAttachments = downloadAttachments;
+    this.context = new AttachmentContext(db, tableName, logger ?? db.logger, archivedCacheLimit);
+    this.attachmentService = new AttachmentService(db, logger ?? db.logger, tableName);
+    this.syncingService = new SyncingService(this.context, localStorage, remoteStorage, logger ?? db.logger, errorHandler);
+    this.watchActiveAttachments = this.attachmentService.watchActiveAttachments({ throttleMs: this.syncThrottleDuration });
   }
 
   /**
@@ -346,6 +346,18 @@ export class AttachmentQueue {
     });
   }
 
+  async expireCache(): Promise<void> {
+    let isDone = false;
+    while (!isDone) {
+      isDone = await this.syncingService.deleteArchivedAttachments();
+    }
+  }
+
+  async clearQueue(): Promise<void> {
+    await this.context.clearQueue();
+    await this.localStorage.clear();
+  }
+
   /**
    * Verifies the integrity of all attachment records and repairs inconsistencies.
    * 
@@ -372,15 +384,23 @@ export class AttachmentQueue {
       const newLocalUri = this.localStorage.getLocalUri(attachment.filename);
       const newExists = await this.localStorage.fileExists(newLocalUri);
       if (newExists) {
-        // The file exists but the localUri is broken, lets update it.
+        // The file exists locally but the localUri is broken, we update it.
         updates.push({
           ...attachment,
           localUri: newLocalUri
         });
       } else {
-        // no new exists
-        if (attachment.state === AttachmentState.QUEUED_UPLOAD || attachment.state === AttachmentState.SYNCED) {
-          // The file must have been removed from the local storage before upload was completed
+        // the file doesn't exist locally.
+        if (attachment.state === AttachmentState.SYNCED) {
+          // the file has been successfully synced to remote storage but is missing
+          // we download it again
+          updates.push({
+            ...attachment,
+            state: AttachmentState.QUEUED_DOWNLOAD,
+            localUri: undefined
+          });
+        } else {
+          // the file wasn't successfully synced to remote storage, we archive it
           updates.push({
             ...attachment,
             state: AttachmentState.ARCHIVED,

packages/attachments/src/AttachmentService.ts

@@ -1,20 +1,23 @@
-import { AbstractPowerSyncDatabase, DifferentialWatchedQuery } from '@powersync/common';
+import { AbstractPowerSyncDatabase, DifferentialWatchedQuery, ILogger } from '@powersync/common';
 import { AttachmentRecord, AttachmentState } from './Schema.js';
+import { AttachmentContext } from './AttachmentContext.js';
 
 /**
  * Service for querying and watching attachment records in the database.
  */
 export class AttachmentService {
   constructor(
+    private db: AbstractPowerSyncDatabase,
+    private logger: ILogger,
     private tableName: string = 'attachments',
-    private db: AbstractPowerSyncDatabase
   ) {}
 
   /**
    * Creates a differential watch query for active attachments requiring synchronization.
    * @returns Watch query that emits changes for queued uploads, downloads, and deletes
    */
   watchActiveAttachments({ throttleMs }: { throttleMs?: number } = {}): DifferentialWatchedQuery<AttachmentRecord> {
+    this.logger.info('Watching active attachments...');
     const watch = this.db
       .query<AttachmentRecord>({
         sql: /* sql */ `

packages/attachments/src/SyncingService.ts

@@ -170,17 +170,18 @@ export class SyncingService {
    * Performs cleanup of archived attachments by removing their local files and records.
    * Errors during local file deletion are logged but do not prevent record deletion.
    */
-  async deleteArchivedAttachments(): Promise<void> {
-    const archivedAttachments = await this.context.getArchivedAttachments();
-    for (const attachment of archivedAttachments) {
-      if (attachment.localUri) {
-        try {
-          await this.localStorage.deleteFile(attachment.localUri);
-        } catch (error) {
-          this.logger.error('Error deleting local file for archived attachment', error);
+  async deleteArchivedAttachments(): Promise<boolean> {
+    return await this.context.deleteArchivedAttachments(async (archivedAttachments) => {
+      for (const attachment of archivedAttachments) {
+        if (attachment.localUri) {
+          try {
+            await this.localStorage.deleteFile(attachment.localUri);
+          } catch (error) {
+            this.logger.error('Error deleting local file for archived attachment', error);
+          }
         }
+        await this.context.deleteAttachment(attachment.id);
       }
-      await this.context.deleteAttachment(attachment.id);
-    }
+    });
   }
 }

packages/attachments/tests/attachments.test.ts

@@ -99,6 +99,7 @@ beforeEach(() => {
     remoteStorage: mockRemoteStorage,
     localStorage: mockLocalStorage,
     syncIntervalMs: INTERVAL_MILLISECONDS,
+    archivedCacheLimit: 0,
   });
 })
 
@@ -280,7 +281,7 @@ describe('attachment queue', () => {
       5
     );
 
-    // await queue.syncStorage(); // <-- explicitly delete
+    // await new Promise(resolve => setTimeout(resolve, 1500));
 
     // File should be deleted too
     expect(await mockLocalStorage.fileExists(record.localUri!)).toBe(false);