add pending actions

2026-08-26 17:29:10 +02:00
parent 664e1dd290
commit f4024c8aae
7 changed files with 423 additions and 17 deletions
@@ -0,0 +1,168 @@
+import { fileStore } from './fileStore';
+import { apiClient } from '../api/client';
+import { ENDPOINTS } from '../constants/api';
+import { HttpError } from '../types';
+import type { PendingAction, PendingActionType } from '../types';
+
+type Listener = () => void;
+
+const MAX_ATTEMPTS = 5;
+const BASE_RETRY_DELAY_MS = 2000;
+
+function genId(): string {
+  return `action_${Date.now()}_${Math.random().toString(36).slice(2, 8)}`;
+}
+
+async function executeAction(action: PendingAction): Promise<void> {
+  const { type, payload, resourceId } = action;
+
+  switch (type) {
+    case 'tag_add': {
+      const { fileId, tags } = payload as { fileId: string; tags: string[] };
+      await apiClient.post(`${ENDPOINTS.RESOURCES}/${fileId}/tags`, { tags });
+      break;
+    }
+    case 'delete': {
+      const { backendId } = payload as { backendId: string };
+      try {
+        await apiClient.delete(`${ENDPOINTS.RESOURCES}/${backendId}`);
+      } catch (err) {
+        if (err instanceof HttpError && err.status === 404) {
+          break;
+        }
+        throw err;
+      }
+      break;
+    }
+    case 'move': {
+      const { resourceIds, parentResourceId } = payload as { resourceIds: string[]; parentResourceId: string | null };
+      await apiClient.post(ENDPOINTS.MOVE, { resource_ids: resourceIds, parent_resource_id: parentResourceId });
+      break;
+    }
+    case 'create_folder': {
+      const { name, parentResourceId } = payload as { name: string; parentResourceId?: string };
+      await apiClient.post(ENDPOINTS.FOLDERS, { name, parent_resource_id: parentResourceId });
+      break;
+    }
+    default:
+      throw new Error(`Unknown action type: ${type}`);
+  }
+}
+
+class ActionQueue {
+  private processing = false;
+  private listeners = new Set<Listener>();
+  private timer: ReturnType<typeof setTimeout> | null = null;
+
+  subscribe(listener: Listener): () => void {
+    this.listeners.add(listener);
+    return () => { this.listeners.delete(listener); };
+  }
+
+  private notify() {
+    this.listeners.forEach((l) => l());
+  }
+
+  enqueue(type: PendingActionType, payload: Record<string, unknown>, resourceId?: string): PendingAction {
+    const action: PendingAction = {
+      id: genId(),
+      type,
+      payload,
+      status: 'pending',
+      attempts: 0,
+      lastError: null,
+      resourceId: resourceId ?? null,
+      createdAt: new Date().toISOString(),
+    };
+    fileStore.insertPendingAction(action);
+    this.notify();
+    return action;
+  }
+
+  async processNext(): Promise<boolean> {
+    const pending = fileStore.getPendingActions();
+    if (pending.length === 0) return false;
+
+    const action = pending[0];
+
+    // Check if a prior delete makes this action obsolete
+    if (action.type !== 'delete' && action.resourceId) {
+      const priorDelete = fileStore.getPendingActions().find(
+        (a) => a.type === 'delete' && a.resourceId === action.resourceId && a.createdAt < action.createdAt
+      );
+      if (priorDelete) {
+        fileStore.markPendingActionObsolete(action.id);
+        this.notify();
+        return true;
+      }
+    }
+
+    try {
+      await executeAction(action);
+      fileStore.markPendingActionDone(action.id);
+      this.notify();
+      return true;
+    } catch (err) {
+      const msg = err instanceof Error ? err.message : String(err);
+      fileStore.markPendingActionError(action.id, msg);
+
+      if (action.attempts + 1 >= MAX_ATTEMPTS) {
+        fileStore.markPendingActionObsolete(action.id);
+      }
+
+      this.notify();
+      return false;
+    }
+  }
+
+  async processAll(): Promise<void> {
+    if (this.processing) return;
+    this.processing = true;
+
+    try {
+      let hasMore = true;
+      while (hasMore) {
+        const pending = fileStore.getPendingActions();
+        if (pending.length === 0) break;
+
+        const success = await this.processNext();
+        if (!success) {
+          // Wait before retrying on error
+          await new Promise((r) => setTimeout(r, BASE_RETRY_DELAY_MS));
+          // Check if there are still pending actions (not just the one we failed on)
+          const remaining = fileStore.getPendingActions();
+          if (remaining.length === 0 || remaining[0].status !== 'pending') break;
+        }
+      }
+    } finally {
+      this.processing = false;
+      fileStore.clearDonePendingActions();
+      this.notify();
+    }
+  }
+
+  isProcessing(): boolean {
+    return this.processing;
+  }
+
+  getPendingCount(): number {
+    return fileStore.getPendingActionsCount();
+  }
+
+  scheduleRetry(delayMs: number = 10_000) {
+    if (this.timer) return;
+    this.timer = setTimeout(() => {
+      this.timer = null;
+      this.processAll();
+    }, delayMs);
+  }
+
+  cancelSchedule() {
+    if (this.timer) {
+      clearTimeout(this.timer);
+      this.timer = null;
+    }
+  }
+}
+
+export const actionQueue = new ActionQueue();
@@ -6,11 +6,13 @@ const BACKGROUND_UPLOAD_TASK = 'BACKGROUND_UPLOAD';
 TaskManager.defineTask(BACKGROUND_UPLOAD_TASK, async () => {
  try {
    const { uploadQueue } = await import('./uploadQueue');
+    const { actionQueue } = await import('./actionQueue');
    const { apiClient } = await import('../api/client');
    const { tokenStorage } = await import('../api/secureStorage');

-    const pending = uploadQueue.getPendingCount();
-    if (pending === 0) {
+    const pendingUploads = uploadQueue.getPendingCount();
+    const pendingActions = actionQueue.getPendingCount();
+    if (pendingUploads === 0 && pendingActions === 0) {
      return BackgroundTask.BackgroundTaskResult.Success;
    }

@@ -20,11 +22,16 @@ TaskManager.defineTask(BACKGROUND_UPLOAD_TASK, async () => {
    }
    apiClient.setAccessToken(token);

-    uploadQueue.retryAll();
+    if (pendingUploads > 0) {
+      uploadQueue.retryAll();
+    }
+    if (pendingActions > 0) {
+      actionQueue.processAll();
+    }

    await new Promise<void>((resolve) => {
      const check = setInterval(() => {
-        if (uploadQueue.getPendingCount() === 0) {
+        if (uploadQueue.getPendingCount() === 0 && actionQueue.getPendingCount() === 0) {
          clearInterval(check);
          resolve();
        }
@@ -1,12 +1,12 @@
 import { drizzle } from 'drizzle-orm/expo-sqlite';
 import * as SQLite from 'expo-sqlite';
 import { eq, like, or, and, desc, asc, sql, isNull } from 'drizzle-orm';
-import { files, fileTags, deletedFiles } from './schema';
-import type { Tag } from '../../types';
+import { files, fileTags, deletedFiles, pendingActions } from './schema';
+import type { Tag, PendingAction, PendingActionType, PendingActionStatus } from '../../types';

 const DB_NAME = 'vaultdrop-v3.db';
 const SCHEMA_VERSION_KEY = 'schema_version';
-const SCHEMA_VERSION = 3;
+const SCHEMA_VERSION = 4;

 let _db: ReturnType<typeof drizzle> | null = null;
 let _sqliteDb: SQLite.SQLiteDatabase | null = null;
@@ -46,6 +46,7 @@ function dropAllTables(db: SQLite.SQLiteDatabase) {
  db.execSync(`DROP TABLE IF EXISTS deleted_files;`);
  db.execSync(`DROP TABLE IF EXISTS device_info;`);
  db.execSync(`DROP TABLE IF EXISTS resources_fts;`);
+  db.execSync(`DROP TABLE IF EXISTS pending_actions;`);
  db.execSync(`DROP TABLE IF EXISTS schema_version;`);
 }

@@ -104,6 +105,21 @@ function createSchema(db: SQLite.SQLiteDatabase) {
    );
  `);

+  db.execSync(`
+    CREATE TABLE IF NOT EXISTS pending_actions (
+      id TEXT PRIMARY KEY,
+      type TEXT NOT NULL,
+      payload TEXT NOT NULL,
+      status TEXT NOT NULL DEFAULT 'pending',
+      attempts INTEGER NOT NULL DEFAULT 0,
+      last_error TEXT,
+      resource_id TEXT,
+      created_at TEXT NOT NULL
+    );
+  `);
+  db.execSync(`CREATE INDEX IF NOT EXISTS idx_pending_actions_status ON pending_actions(status);`);
+  db.execSync(`CREATE INDEX IF NOT EXISTS idx_pending_actions_type ON pending_actions(type);`);
+
  db.execSync(`
    CREATE VIRTUAL TABLE IF NOT EXISTS resources_fts USING fts5(
      name,
@@ -598,4 +614,111 @@ export const fileStore = {
  resetSyncError(id: string) {
    this.updatePartial(id, { syncStatus: 'local' });
  },
+
+  // --- Pending Actions ---
+
+  insertPendingAction(action: PendingAction) {
+    const d = getDb();
+    d.insert(pendingActions).values({
+      id: action.id,
+      type: action.type,
+      payload: JSON.stringify(action.payload),
+      status: action.status,
+      attempts: action.attempts,
+      lastError: action.lastError,
+      resourceId: action.resourceId,
+      createdAt: action.createdAt,
+    }).run();
+  },
+
+  getPendingActions(): PendingAction[] {
+    const d = getDb();
+    const rows = d.select().from(pendingActions)
+      .where(eq(pendingActions.status, 'pending'))
+      .orderBy(asc(pendingActions.createdAt))
+      .all();
+    return rows.map(rowToAction);
+  },
+
+  getPendingActionById(id: string): PendingAction | null {
+    const d = getDb();
+    const row = d.select().from(pendingActions)
+      .where(eq(pendingActions.id, id))
+      .get();
+    return row ? rowToAction(row) : null;
+  },
+
+  getPendingActionsCount(): number {
+    const d = getDb();
+    const row = d.select({ count: sql<number>`count(*)` })
+      .from(pendingActions)
+      .where(eq(pendingActions.status, 'pending'))
+      .get();
+    return row?.count ?? 0;
+  },
+
+  markPendingActionDone(id: string) {
+    const d = getDb();
+    d.update(pendingActions)
+      .set({ status: 'done' })
+      .where(eq(pendingActions.id, id))
+      .run();
+  },
+
+  markPendingActionError(id: string, error: string) {
+    const d = getDb();
+    const row = d.select().from(pendingActions)
+      .where(eq(pendingActions.id, id))
+      .get();
+    if (!row) return;
+    d.update(pendingActions)
+      .set({
+        status: 'error',
+        attempts: row.attempts + 1,
+        lastError: error,
+      })
+      .where(eq(pendingActions.id, id))
+      .run();
+  },
+
+  markPendingActionObsolete(id: string) {
+    const d = getDb();
+    d.update(pendingActions)
+      .set({ status: 'obsolete' })
+      .where(eq(pendingActions.id, id))
+      .run();
+  },
+
+  deletePendingAction(id: string) {
+    const d = getDb();
+    d.delete(pendingActions).where(eq(pendingActions.id, id)).run();
+  },
+
+  clearDonePendingActions() {
+    const d = getDb();
+    d.delete(pendingActions).where(eq(pendingActions.status, 'done')).run();
+    d.delete(pendingActions).where(eq(pendingActions.status, 'obsolete')).run();
+  },
 };
+
+function rowToAction(row: {
+  id: string;
+  type: string;
+  payload: string;
+  status: string;
+  attempts: number;
+  lastError: string | null;
+  resourceId: string | null;
+  createdAt: string;
+}): PendingAction {
+  return {
+    id: row.id,
+    type: row.type as PendingActionType,
+    payload: JSON.parse(row.payload),
+    status: row.status as PendingActionStatus,
+    attempts: row.attempts,
+    lastError: row.lastError,
+    resourceId: row.resourceId,
+    createdAt: row.createdAt,
+  };
+}
@@ -54,3 +54,21 @@ export const deviceInfo = sqliteTable('device_info', {
  platform: text('platform').notNull().default(''),
  registeredAt: text('registered_at'),
 });
+
+export const pendingActions = sqliteTable(
+  'pending_actions',
+  {
+    id: text('id').primaryKey(),
+    type: text('type').notNull(),
+    payload: text('payload').notNull(),
+    status: text('status').notNull().default('pending'),
+    attempts: integer('attempts').notNull().default(0),
+    lastError: text('last_error'),
+    resourceId: text('resource_id'),
+    createdAt: text('created_at').notNull(),
+  },
+  (t) => [
+    index('idx_pending_actions_status').on(t.status),
+    index('idx_pending_actions_type').on(t.type),
+  ],
+);