Refactor watch active observer into dedicated service

Author: joshuabrink

packages/attachments/src/AttachmentContext.ts

@@ -12,36 +12,6 @@ export class AttachmentContext {
     this.logger = logger;
   }
 
-  watchActiveAttachments(onUpdate: () => void): AbortController {
-    const abortController = new AbortController();
-    this.db.watchWithCallback(
-      /* sql */
-      `
-        SELECT
-          *
-        FROM
-          ${this.tableName}
-        WHERE
-          state = ?
-          OR state = ?
-          OR state = ?
-        ORDER BY
-          timestamp ASC
-      `,
-      [AttachmentState.QUEUED_UPLOAD, AttachmentState.QUEUED_DOWNLOAD, AttachmentState.QUEUED_DELETE],
-      {
-        onResult: () => {
-          onUpdate();
-        }
-      },
-      {
-        signal: abortController.signal
-      }
-    );
-
-    return abortController;
-  }
-
   async getActiveAttachments(): Promise<any[]> {
     const attachments = await this.db.getAll(
       /* sql */

packages/attachments/src/AttachmentQueue.ts

@@ -5,6 +5,7 @@ import { RemoteStorageAdapter } from './RemoteStorageAdapter.js';
 import { ATTACHMENT_TABLE, AttachmentRecord, AttachmentState } from './Schema.js';
 import { StorageService } from './StorageService.js';
 import { WatchedAttachmentItem } from './WatchedAttachmentItem.js';
+import { AttachmentService } from './AttachmentService.js';
 
 export class AttachmentQueue {
   periodicSyncTimer?: ReturnType<typeof setInterval>;
@@ -20,6 +21,7 @@ export class AttachmentQueue {
   downloadAttachments: boolean = true;
   watchActiveAbortController?: AbortController;
   archivedCacheLimit: number;
+  attachmentService: AttachmentService;
 
   constructor({
     db,
@@ -51,6 +53,7 @@ export class AttachmentQueue {
     this.tableName = tableName;
     this.storageService = new StorageService(this.context, localStorage, remoteStorage, logger ?? db.logger);
     this.syncInterval = syncInterval;
+    this.attachmentService = new AttachmentService(tableName, db);
     this.syncThrottleDuration = syncThrottleDuration;
     this.downloadAttachments = downloadAttachments;
     this.archivedCacheLimit = archivedCacheLimit;
@@ -69,8 +72,10 @@ export class AttachmentQueue {
     }, this.syncInterval);
 
     // Sync storage when there is a change in active attachments
-    this.watchActiveAbortController = this.context.watchActiveAttachments(async () => {
-      await this.syncStorage();
+    this.attachmentService.watchActiveAttachments().registerListener({
+      onDiff: async () => {
+        await this.syncStorage();
+      }
     });
 
     // Process attachments when there is a change in watched attachments
@@ -166,7 +171,7 @@ export class AttachmentQueue {
   async stopSync(): Promise<void> {
     clearInterval(this.periodicSyncTimer);
     this.periodicSyncTimer = undefined;
-    this.watchActiveAbortController?.abort();
+    await this.attachmentService.watchActiveAttachments().close();
   }
 
   async saveFile({

packages/attachments/src/AttachmentService.ts

@@ -0,0 +1,38 @@
+import { AbstractPowerSyncDatabase, DifferentialWatchedQuery } from '@powersync/common';
+import { AttachmentRecord, AttachmentState } from './Schema.js';
+
+/**
+ * Service for querying and watching attachment records in the database.
+ */
+export class AttachmentService {
+  constructor(
+    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(): DifferentialWatchedQuery<AttachmentRecord> {
+    const watch = this.db
+      .query<AttachmentRecord>({
+        sql: /* sql */ `
+          SELECT
+            *
+          FROM
+            ${this.tableName}
+          WHERE
+            state = ?
+            OR state = ?
+            OR state = ?
+          ORDER BY
+            timestamp ASC
+        `,
+        parameters: [AttachmentState.QUEUED_UPLOAD, AttachmentState.QUEUED_DOWNLOAD, AttachmentState.QUEUED_DELETE]
+      })
+      .differentialWatch();
+
+    return watch;
+  }
+}