From f4024c8aae2f345c3badc157ad85682fc4a5e113 Mon Sep 17 00:00:00 2001 From: m Date: Wed, 26 Aug 2026 17:29:10 +0200 Subject: [PATCH] add pending actions --- mobile/hooks/useFiles.ts | 59 ++++++++-- mobile/hooks/usePendingActions.ts | 37 ++++++ mobile/services/actionQueue.ts | 168 ++++++++++++++++++++++++++++ mobile/services/backgroundUpload.ts | 15 ++- mobile/services/fileStore/index.ts | 129 ++++++++++++++++++++- mobile/services/fileStore/schema.ts | 18 +++ mobile/types/index.ts | 14 +++ 7 files changed, 423 insertions(+), 17 deletions(-) create mode 100644 mobile/hooks/usePendingActions.ts create mode 100644 mobile/services/actionQueue.ts diff --git a/mobile/hooks/useFiles.ts b/mobile/hooks/useFiles.ts index d69d1b3..76bdb1d 100644 --- a/mobile/hooks/useFiles.ts +++ b/mobile/hooks/useFiles.ts @@ -3,6 +3,7 @@ import { apiClient } from '../api/client'; import { ENDPOINTS } from '../constants/api'; import type { UnifiedFileItem, PaginatedResponse, FileItem, Tag } from '../types'; import { fileStore } from '../services/fileStore'; +import { actionQueue } from '../services/actionQueue'; function recordToUnifiedItem(record: ReturnType): UnifiedFileItem | null { if (!record) return null; @@ -138,9 +139,12 @@ export function useDeleteFile() { return useMutation({ mutationFn: async (id: string) => { - const result = await apiClient.delete(`${ENDPOINTS.RESOURCES}/${id}`); + const record = fileStore.getByBackendId(id) ?? fileStore.getById(id); + + if (record?.backendId) { + actionQueue.enqueue('delete', { backendId: record.backendId }, record.backendId); + } fileStore.deleteByBackendId(id); - return result; }, onSuccess: () => { queryClient.invalidateQueries({ queryKey: ['resources'] }); @@ -152,8 +156,17 @@ export function useAddTags() { const queryClient = useQueryClient(); return useMutation({ - mutationFn: ({ fileId, tags }: { fileId: string; tags: string[] }) => - apiClient.post(`${ENDPOINTS.RESOURCES}/${fileId}/tags`, { tags }), + mutationFn: ({ fileId, tags }: { fileId: string; tags: string[] }) => { + const record = fileStore.getById(fileId) ?? fileStore.getByBackendId(fileId); + if (record) { + const existingTags = record.tags ?? []; + const newTags = [...existingTags, ...tags.map((t) => ({ id: t, tag_name: t }))]; + const uniqueTags = newTags.filter((t, i, arr) => arr.findIndex((x) => x.tag_name === t.tag_name) === i); + fileStore.updatePartial(record.id, {}); + actionQueue.enqueue('tag_add', { fileId: record.backendId ?? record.id, tags }, record.backendId ?? record.id); + } + return Promise.resolve(); + }, onSuccess: () => { queryClient.invalidateQueries({ queryKey: ['resources'] }); }, @@ -164,15 +177,21 @@ export function useMoveResources() { const queryClient = useQueryClient(); return useMutation({ - mutationFn: ({ resourceIds, parentResourceId }: { resourceIds: string[]; parentResourceId: string | null }) => - apiClient.post(ENDPOINTS.MOVE, { resource_ids: resourceIds, parent_resource_id: parentResourceId }), - onSuccess: (_, { resourceIds, parentResourceId }) => { + mutationFn: ({ resourceIds, parentResourceId }: { resourceIds: string[]; parentResourceId: string | null }) => { + const backendIds: string[] = []; for (const id of resourceIds) { const record = fileStore.getById(id) ?? fileStore.getByBackendId(id); - if (record) { + if (record?.backendId) { + backendIds.push(record.backendId); fileStore.updatePartial(record.id, { parentResourceId: parentResourceId ?? null }); } } + if (backendIds.length > 0) { + actionQueue.enqueue('move', { resourceIds: backendIds, parentResourceId }); + } + return Promise.resolve(); + }, + onSuccess: () => { queryClient.invalidateQueries({ queryKey: ['resources'] }); }, }); @@ -214,8 +233,28 @@ export function useCreateFolder() { return useMutation({ mutationFn: async ({ name, parentResourceId }: { name: string; parentResourceId?: string }) => { - const res = await apiClient.post<{ data: FileItem }>(ENDPOINTS.FOLDERS, { name, parent_resource_id: parentResourceId }); - return res.data; + const now = new Date().toISOString(); + const localId = `local_folder_${Date.now()}_${Math.random().toString(36).slice(2, 8)}`; + fileStore.upsert({ + id: localId, + backendId: null, + name, + mimeType: 'inode/directory', + size: 0, + source: 'local', + localUri: null, + syncStatus: 'local', + parentResourceId: parentResourceId ?? null, + isFolder: 1, + ocrText: null, + thumbnailUrl: null, + ownerId: null, + createdAt: now, + updatedAt: now, + lastSyncedAt: null, + }); + actionQueue.enqueue('create_folder', { name, parentResourceId }, localId); + return { id: localId, name }; }, onSuccess: () => { queryClient.invalidateQueries({ queryKey: ['resources'] }); diff --git a/mobile/hooks/usePendingActions.ts b/mobile/hooks/usePendingActions.ts new file mode 100644 index 0000000..f73348a --- /dev/null +++ b/mobile/hooks/usePendingActions.ts @@ -0,0 +1,37 @@ +import { useState, useEffect, useCallback, useSyncExternalStore, useRef } from 'react'; +import { actionQueue } from '../services/actionQueue'; +import { useNetworkStatus } from './useNetworkStatus'; + +function subscribe(listener: () => void): () => void { + return actionQueue.subscribe(listener); +} + +function getSnapshot(): number { + return actionQueue.getPendingCount(); +} + +export function usePendingActions() { + const pendingCount = useSyncExternalStore(subscribe, getSnapshot); + const { isOnline } = useNetworkStatus(); + const wasOffline = useRef(!isOnline); + + useEffect(() => { + if (isOnline && wasOffline.current && pendingCount > 0) { + actionQueue.processAll(); + } + wasOffline.current = !isOnline; + }, [isOnline, pendingCount]); + + useEffect(() => { + if (pendingCount > 0 && isOnline) { + actionQueue.scheduleRetry(5_000); + } + return () => actionQueue.cancelSchedule(); + }, [pendingCount, isOnline]); + + return { + pendingCount, + isProcessing: actionQueue.isProcessing(), + processAll: useCallback(() => actionQueue.processAll(), []), + }; +} diff --git a/mobile/services/actionQueue.ts b/mobile/services/actionQueue.ts new file mode 100644 index 0000000..25456e6 --- /dev/null +++ b/mobile/services/actionQueue.ts @@ -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 { + 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(); + private timer: ReturnType | 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, 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 { + 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 { + 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(); diff --git a/mobile/services/backgroundUpload.ts b/mobile/services/backgroundUpload.ts index d23baa1..64dc174 100644 --- a/mobile/services/backgroundUpload.ts +++ b/mobile/services/backgroundUpload.ts @@ -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((resolve) => { const check = setInterval(() => { - if (uploadQueue.getPendingCount() === 0) { + if (uploadQueue.getPendingCount() === 0 && actionQueue.getPendingCount() === 0) { clearInterval(check); resolve(); } diff --git a/mobile/services/fileStore/index.ts b/mobile/services/fileStore/index.ts index f23e3b0..de41a9d 100644 --- a/mobile/services/fileStore/index.ts +++ b/mobile/services/fileStore/index.ts @@ -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 | 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`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, + }; +} diff --git a/mobile/services/fileStore/schema.ts b/mobile/services/fileStore/schema.ts index c445717..cc1da2f 100644 --- a/mobile/services/fileStore/schema.ts +++ b/mobile/services/fileStore/schema.ts @@ -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), + ], +); diff --git a/mobile/types/index.ts b/mobile/types/index.ts index 8959692..899b84c 100644 --- a/mobile/types/index.ts +++ b/mobile/types/index.ts @@ -164,3 +164,17 @@ export interface ResourcePlacement { storageKey: string | null; syncedAt: string | null; } + +export type PendingActionType = 'tag_add' | 'delete' | 'move' | 'create_folder'; +export type PendingActionStatus = 'pending' | 'done' | 'error' | 'obsolete'; + +export interface PendingAction { + id: string; + type: PendingActionType; + payload: Record; + status: PendingActionStatus; + attempts: number; + lastError: string | null; + resourceId: string | null; + createdAt: string; +}