|
|
|
@@ -1,784 +0,0 @@
|
|
|
|
|
# Import Queue Implementation Plan
|
|
|
|
|
|
|
|
|
|
> **For Claude:** REQUIRED SUB-SKILL: Use superpowers:executing-plans to implement this plan task-by-task.
|
|
|
|
|
|
|
|
|
|
**Goal:** Replace the synchronous fire-and-wait import with a SQLite-backed job queue so imports survive page closes and show live progress via polling.
|
|
|
|
|
|
|
|
|
|
**Architecture:** A job is created in SQLite on request, a background worker processes items concurrently (4 at a time) and writes per-item status back to the db. The frontend polls a status endpoint every 2s. Platform-specific logic is encapsulated in processor classes so Nextcloud/PhotoPrism can be added by dropping in a new file. On server restart, any stuck `processing` jobs are reset to `pending` and restarted when the user's next poll hits the status endpoint.
|
|
|
|
|
|
|
|
|
|
**Tech Stack:** Drizzle ORM + libsql (SQLite), Next.js API routes, TypeScript, React (polling via `useEffect` + `setInterval`)
|
|
|
|
|
|
|
|
|
|
---
|
|
|
|
|
|
|
|
|
|
### Task 1: Add schema tables
|
|
|
|
|
|
|
|
|
|
**Files:**
|
|
|
|
|
- Create: `src/db/schema/importJobs.schema.ts`
|
|
|
|
|
- Modify: `src/db/schema/index.ts`
|
|
|
|
|
|
|
|
|
|
**Step 1: Create the schema file**
|
|
|
|
|
|
|
|
|
|
```typescript
|
|
|
|
|
// src/db/schema/importJobs.schema.ts
|
|
|
|
|
import { sqliteTable, text, integer } from "drizzle-orm/sqlite-core";
|
|
|
|
|
import { randomUUID } from "crypto";
|
|
|
|
|
|
|
|
|
|
export const importJobs = sqliteTable("import_jobs", {
|
|
|
|
|
id: text("id").primaryKey().$defaultFn(() => randomUUID()),
|
|
|
|
|
userId: text("user_id").notNull(),
|
|
|
|
|
platform: text("platform").notNull(), // 'immich' | 'nextcloud' | 'photoprism'
|
|
|
|
|
status: text("status").notNull().default("pending"), // pending | processing | completed | failed
|
|
|
|
|
url: text("url").notNull(), // base URL of source
|
|
|
|
|
urlConfig: text("url_config").notNull().default("{}"), // JSON: auth/connection (key, token, etc.)
|
|
|
|
|
importData: text("import_data").notNull().default("{}"), // JSON: album options, resolved albumId, etc.
|
|
|
|
|
totalCount: integer("total_count").notNull().default(0),
|
|
|
|
|
uploadedCount: integer("uploaded_count").notNull().default(0),
|
|
|
|
|
skippedCount: integer("skipped_count").notNull().default(0),
|
|
|
|
|
failedCount: integer("failed_count").notNull().default(0),
|
|
|
|
|
createdAt: integer("created_at", { mode: "timestamp" }).$defaultFn(() => new Date()),
|
|
|
|
|
updatedAt: integer("updated_at", { mode: "timestamp" })
|
|
|
|
|
.$defaultFn(() => new Date())
|
|
|
|
|
.$onUpdate(() => new Date()),
|
|
|
|
|
});
|
|
|
|
|
|
|
|
|
|
export const importJobItems = sqliteTable("import_job_items", {
|
|
|
|
|
id: text("id").primaryKey().$defaultFn(() => randomUUID()),
|
|
|
|
|
jobId: text("job_id").notNull().references(() => importJobs.id),
|
|
|
|
|
assetId: text("asset_id").notNull(), // source platform's asset ID
|
|
|
|
|
status: text("status").notNull().default("pending"), // pending | processing | uploaded | skipped | failed
|
|
|
|
|
itemData: text("item_data").notNull().default("{}"), // JSON: filename, type, dates, isFavorite, etc.
|
|
|
|
|
immichId: text("immich_id"), // Immich asset ID after successful upload
|
|
|
|
|
error: text("error"),
|
|
|
|
|
createdAt: integer("created_at", { mode: "timestamp" }).$defaultFn(() => new Date()),
|
|
|
|
|
updatedAt: integer("updated_at", { mode: "timestamp" })
|
|
|
|
|
.$defaultFn(() => new Date())
|
|
|
|
|
.$onUpdate(() => new Date()),
|
|
|
|
|
});
|
|
|
|
|
```
|
|
|
|
|
|
|
|
|
|
**Step 2: Export from schema index**
|
|
|
|
|
|
|
|
|
|
In `src/db/schema/index.ts`, add:
|
|
|
|
|
```typescript
|
|
|
|
|
export * from "./importJobs.schema";
|
|
|
|
|
```
|
|
|
|
|
|
|
|
|
|
**Step 3: Commit**
|
|
|
|
|
|
|
|
|
|
```bash
|
|
|
|
|
git add src/db/schema/importJobs.schema.ts src/db/schema/index.ts
|
|
|
|
|
git commit -m "feat: add import_jobs and import_job_items schema tables"
|
|
|
|
|
```
|
|
|
|
|
|
|
|
|
|
---
|
|
|
|
|
|
|
|
|
|
### Task 2: Generate and apply migration
|
|
|
|
|
|
|
|
|
|
**Files:**
|
|
|
|
|
- Create: `src/db/migrations/` (auto-generated by drizzle-kit)
|
|
|
|
|
|
|
|
|
|
**Step 1: Generate migration**
|
|
|
|
|
|
|
|
|
|
```bash
|
|
|
|
|
bun run db:generate
|
|
|
|
|
```
|
|
|
|
|
|
|
|
|
|
Expected: a new `.sql` file appears in `src/db/migrations/` containing `CREATE TABLE import_jobs` and `CREATE TABLE import_job_items`.
|
|
|
|
|
|
|
|
|
|
**Step 2: Verify migration SQL looks correct**
|
|
|
|
|
|
|
|
|
|
Open the generated file and confirm it has both tables with all columns. No manual edits needed.
|
|
|
|
|
|
|
|
|
|
**Step 3: Test that migrations run**
|
|
|
|
|
|
|
|
|
|
```bash
|
|
|
|
|
bun run dev
|
|
|
|
|
```
|
|
|
|
|
|
|
|
|
|
Expected: server starts without errors. The migration auto-runs via `instrumentation.ts → runMigrations()`.
|
|
|
|
|
|
|
|
|
|
**Step 4: Commit**
|
|
|
|
|
|
|
|
|
|
```bash
|
|
|
|
|
git add src/db/migrations/
|
|
|
|
|
git commit -m "feat: add migration for import_jobs and import_job_items"
|
|
|
|
|
```
|
|
|
|
|
|
|
|
|
|
---
|
|
|
|
|
|
|
|
|
|
### Task 3: Processor interface and registry
|
|
|
|
|
|
|
|
|
|
**Files:**
|
|
|
|
|
- Create: `src/workers/import/types.ts`
|
|
|
|
|
- Create: `src/workers/import/registry.ts`
|
|
|
|
|
|
|
|
|
|
**Step 1: Define the processor interface**
|
|
|
|
|
|
|
|
|
|
```typescript
|
|
|
|
|
// src/workers/import/types.ts
|
|
|
|
|
import type { importJobs, importJobItems } from "@/db/schema";
|
|
|
|
|
|
|
|
|
|
export type ImportJob = typeof importJobs.$inferSelect;
|
|
|
|
|
export type ImportJobItem = typeof importJobItems.$inferSelect;
|
|
|
|
|
|
|
|
|
|
export interface ProcessorContext {
|
|
|
|
|
headers: Record<string, string>; // user auth headers for Immich API calls
|
|
|
|
|
albumId?: string; // resolved after setup()
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
export interface SetupResult {
|
|
|
|
|
skipAssetIds: string[]; // source asset IDs already in Immich → mark skipped
|
|
|
|
|
albumId?: string; // created or resolved album ID
|
|
|
|
|
importDataPatch?: Record<string, unknown>; // merged into job.importData in db
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
export interface ImportProcessor {
|
|
|
|
|
/** Called once per job before items are processed. Dedup + album creation. */
|
|
|
|
|
setup(job: ImportJob, context: ProcessorContext): Promise<SetupResult>;
|
|
|
|
|
/** Called once per item. Returns the Immich asset ID on success, throws on failure. */
|
|
|
|
|
processItem(job: ImportJob, item: ImportJobItem, context: ProcessorContext): Promise<{ immichId: string }>;
|
|
|
|
|
}
|
|
|
|
|
```
|
|
|
|
|
|
|
|
|
|
**Step 2: Create the registry**
|
|
|
|
|
|
|
|
|
|
```typescript
|
|
|
|
|
// src/workers/import/registry.ts
|
|
|
|
|
import type { ImportProcessor } from "./types";
|
|
|
|
|
|
|
|
|
|
const processors = new Map<string, ImportProcessor>();
|
|
|
|
|
|
|
|
|
|
export function registerProcessor(platform: string, processor: ImportProcessor) {
|
|
|
|
|
processors.set(platform, processor);
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
export function getProcessor(platform: string): ImportProcessor {
|
|
|
|
|
const processor = processors.get(platform);
|
|
|
|
|
if (!processor) {
|
|
|
|
|
throw new Error(`No import processor registered for platform: ${platform}`);
|
|
|
|
|
}
|
|
|
|
|
return processor;
|
|
|
|
|
}
|
|
|
|
|
```
|
|
|
|
|
|
|
|
|
|
**Step 3: Commit**
|
|
|
|
|
|
|
|
|
|
```bash
|
|
|
|
|
git add src/workers/import/types.ts src/workers/import/registry.ts
|
|
|
|
|
git commit -m "feat: add import processor interface and registry"
|
|
|
|
|
```
|
|
|
|
|
|
|
|
|
|
---
|
|
|
|
|
|
|
|
|
|
### Task 4: Immich shared-link processor
|
|
|
|
|
|
|
|
|
|
**Files:**
|
|
|
|
|
- Create: `src/workers/import/processors/immich-shared-link.ts`
|
|
|
|
|
|
|
|
|
|
This extracts the core download/upload logic from `src/pages/api/import-shared/upload-all.ts` into the processor interface. The helpers (`downloadSharedAsset`, `uploadAssetBuffer`, `createImmichAlbum`, `addAssetToAlbum`, `ensurePowerToolsTag`) already exist in `upload-all.ts` and `helpers.ts` — copy/import them as needed.
|
|
|
|
|
|
|
|
|
|
**Step 1: Create the processor**
|
|
|
|
|
|
|
|
|
|
```typescript
|
|
|
|
|
// src/workers/import/processors/immich-shared-link.ts
|
|
|
|
|
import { ENV } from "@/config/environment";
|
|
|
|
|
import {
|
|
|
|
|
ensurePowerToolsTag,
|
|
|
|
|
tagAssetWithPowerTools,
|
|
|
|
|
type HeadersRecord,
|
|
|
|
|
} from "@/pages/api/import-shared/helpers";
|
|
|
|
|
import type { ImportJob, ImportJobItem, ImportProcessor, ProcessorContext, SetupResult } from "../types";
|
|
|
|
|
|
|
|
|
|
// --- helpers copied/imported from upload-all.ts ---
|
|
|
|
|
// (parseFileNameFromDisposition, guessContentType, inferAssetTypeFromName,
|
|
|
|
|
// downloadSharedAsset, uploadAssetBuffer, createImmichAlbum, addAssetToAlbum)
|
|
|
|
|
// Copy them here verbatim from upload-all.ts so upload-all.ts stays intact.
|
|
|
|
|
|
|
|
|
|
const DEVICE_ID = "immich-power-tools";
|
|
|
|
|
const makeDeviceAssetId = (assetId: string) => `shared-${assetId}`;
|
|
|
|
|
|
|
|
|
|
export class ImmichSharedLinkProcessor implements ImportProcessor {
|
|
|
|
|
async setup(job: ImportJob, context: ProcessorContext): Promise<SetupResult> {
|
|
|
|
|
const urlConfig = JSON.parse(job.urlConfig) as { key: string };
|
|
|
|
|
const items = await getJobItems(job.id); // helper defined in runner.ts, passed via closure or import
|
|
|
|
|
const deviceAssetIds = items.map((i) => makeDeviceAssetId(i.assetId));
|
|
|
|
|
const lookup = new Map(items.map((i) => [makeDeviceAssetId(i.assetId).toLowerCase(), i.assetId]));
|
|
|
|
|
|
|
|
|
|
const skipAssetIds: string[] = [];
|
|
|
|
|
try {
|
|
|
|
|
const res = await fetch(`${ENV.IMMICH_URL}/api/assets/exist`, {
|
|
|
|
|
method: "POST",
|
|
|
|
|
headers: { ...context.headers, "Content-Type": "application/json" },
|
|
|
|
|
body: JSON.stringify({ deviceAssetIds, deviceId: DEVICE_ID }),
|
|
|
|
|
});
|
|
|
|
|
if (res.ok) {
|
|
|
|
|
const { existingIds = [] } = await res.json();
|
|
|
|
|
for (const id of existingIds) {
|
|
|
|
|
const assetId = lookup.get(id.toLowerCase());
|
|
|
|
|
if (assetId) skipAssetIds.push(assetId);
|
|
|
|
|
}
|
|
|
|
|
}
|
|
|
|
|
} catch (e) {
|
|
|
|
|
console.warn("Dedup check failed, continuing without it", e);
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
const importData = JSON.parse(job.importData) as {
|
|
|
|
|
albumOptions?: { createAlbum?: boolean; albumName?: string; addToAlbumId?: string };
|
|
|
|
|
albumId?: string;
|
|
|
|
|
};
|
|
|
|
|
|
|
|
|
|
let albumId: string | undefined;
|
|
|
|
|
const albumOptions = importData.albumOptions;
|
|
|
|
|
if (albumOptions?.createAlbum) {
|
|
|
|
|
const name = albumOptions.albumName?.trim() || `Shared import ${new Date().toISOString()}`;
|
|
|
|
|
albumId = await createImmichAlbum(name, {
|
|
|
|
|
...context.headers,
|
|
|
|
|
"Content-Type": "application/json",
|
|
|
|
|
} as HeadersRecord);
|
|
|
|
|
} else if (albumOptions?.addToAlbumId) {
|
|
|
|
|
albumId = albumOptions.addToAlbumId;
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
return { skipAssetIds, albumId, importDataPatch: albumId ? { albumId } : undefined };
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
async processItem(job: ImportJob, item: ImportJobItem, context: ProcessorContext): Promise<{ immichId: string }> {
|
|
|
|
|
const urlConfig = JSON.parse(job.urlConfig) as { key: string };
|
|
|
|
|
const itemData = JSON.parse(item.itemData) as {
|
|
|
|
|
originalFileName?: string;
|
|
|
|
|
type: string;
|
|
|
|
|
fileCreatedAt?: string | null;
|
|
|
|
|
localDateTime?: string | null;
|
|
|
|
|
duration?: string | null;
|
|
|
|
|
isFavorite?: boolean;
|
|
|
|
|
isArchived?: boolean;
|
|
|
|
|
};
|
|
|
|
|
|
|
|
|
|
const asset = { id: item.assetId, ...itemData };
|
|
|
|
|
const { ["Content-Type"]: _omit, ...uploadHeaders } = context.headers as HeadersRecord;
|
|
|
|
|
|
|
|
|
|
const downloaded = await downloadSharedAsset(asset, job.url, urlConfig.key);
|
|
|
|
|
const tag = await ensurePowerToolsTag(context.headers as HeadersRecord);
|
|
|
|
|
const immichId = await uploadAssetBuffer(
|
|
|
|
|
asset,
|
|
|
|
|
downloaded,
|
|
|
|
|
uploadHeaders as HeadersRecord,
|
|
|
|
|
context.headers as HeadersRecord,
|
|
|
|
|
tag.id
|
|
|
|
|
);
|
|
|
|
|
|
|
|
|
|
if (context.albumId) {
|
|
|
|
|
await addAssetToAlbum(context.albumId, immichId, {
|
|
|
|
|
...context.headers,
|
|
|
|
|
"Content-Type": "application/json",
|
|
|
|
|
} as HeadersRecord);
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
return { immichId };
|
|
|
|
|
}
|
|
|
|
|
}
|
|
|
|
|
```
|
|
|
|
|
|
|
|
|
|
**Step 2: Register the processor**
|
|
|
|
|
|
|
|
|
|
Create `src/workers/import/processors/index.ts`:
|
|
|
|
|
```typescript
|
|
|
|
|
import { registerProcessor } from "../registry";
|
|
|
|
|
import { ImmichSharedLinkProcessor } from "./immich-shared-link";
|
|
|
|
|
|
|
|
|
|
registerProcessor("immich", new ImmichSharedLinkProcessor());
|
|
|
|
|
```
|
|
|
|
|
|
|
|
|
|
**Step 3: Commit**
|
|
|
|
|
|
|
|
|
|
```bash
|
|
|
|
|
git add src/workers/import/processors/
|
|
|
|
|
git commit -m "feat: add immich shared-link import processor"
|
|
|
|
|
```
|
|
|
|
|
|
|
|
|
|
---
|
|
|
|
|
|
|
|
|
|
### Task 5: Worker runner
|
|
|
|
|
|
|
|
|
|
**Files:**
|
|
|
|
|
- Create: `src/workers/import/runner.ts`
|
|
|
|
|
|
|
|
|
|
The runner owns the active-job tracking (in-memory Set), processes items concurrently, and updates the db after each item.
|
|
|
|
|
|
|
|
|
|
**Step 1: Create the runner**
|
|
|
|
|
|
|
|
|
|
```typescript
|
|
|
|
|
// src/workers/import/runner.ts
|
|
|
|
|
import { eq, inArray } from "drizzle-orm";
|
|
|
|
|
import { appDb } from "@/db";
|
|
|
|
|
import { importJobs, importJobItems } from "@/db/schema";
|
|
|
|
|
import { getProcessor } from "./registry";
|
|
|
|
|
import "./processors/index"; // side-effect: registers all processors
|
|
|
|
|
|
|
|
|
|
export type { ImportJob, ImportJobItem } from "./types";
|
|
|
|
|
|
|
|
|
|
const activeJobs = new Set<string>();
|
|
|
|
|
const CONCURRENCY = 4;
|
|
|
|
|
|
|
|
|
|
export function isJobActive(jobId: string) {
|
|
|
|
|
return activeJobs.has(jobId);
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
export async function getJobItems(jobId: string) {
|
|
|
|
|
return appDb.select().from(importJobItems).where(eq(importJobItems.jobId, jobId));
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
export async function resumePendingJobs() {
|
|
|
|
|
// Reset any jobs stuck in 'processing' from a previous crashed run
|
|
|
|
|
await appDb
|
|
|
|
|
.update(importJobs)
|
|
|
|
|
.set({ status: "pending" })
|
|
|
|
|
.where(eq(importJobs.status, "processing"));
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
export async function runImportJob(
|
|
|
|
|
jobId: string,
|
|
|
|
|
headers: Record<string, string>
|
|
|
|
|
): Promise<void> {
|
|
|
|
|
if (activeJobs.has(jobId)) return;
|
|
|
|
|
activeJobs.add(jobId);
|
|
|
|
|
|
|
|
|
|
try {
|
|
|
|
|
const [job] = await appDb.select().from(importJobs).where(eq(importJobs.id, jobId));
|
|
|
|
|
if (!job || job.status === "completed" || job.status === "failed") return;
|
|
|
|
|
|
|
|
|
|
await appDb.update(importJobs).set({ status: "processing" }).where(eq(importJobs.id, jobId));
|
|
|
|
|
|
|
|
|
|
const processor = getProcessor(job.platform);
|
|
|
|
|
const context = { headers };
|
|
|
|
|
|
|
|
|
|
// --- Setup: dedup + album creation ---
|
|
|
|
|
const { skipAssetIds, albumId, importDataPatch } = await processor.setup(job, context);
|
|
|
|
|
|
|
|
|
|
if (importDataPatch) {
|
|
|
|
|
const merged = { ...JSON.parse(job.importData), ...importDataPatch };
|
|
|
|
|
await appDb.update(importJobs).set({ importData: JSON.stringify(merged) }).where(eq(importJobs.id, jobId));
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
if (skipAssetIds.length > 0) {
|
|
|
|
|
await appDb
|
|
|
|
|
.update(importJobItems)
|
|
|
|
|
.set({ status: "skipped" })
|
|
|
|
|
.where(inArray(importJobItems.assetId, skipAssetIds));
|
|
|
|
|
await appDb
|
|
|
|
|
.update(importJobs)
|
|
|
|
|
.set({ skippedCount: skipAssetIds.length })
|
|
|
|
|
.where(eq(importJobs.id, jobId));
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
const processorContext = { ...context, albumId };
|
|
|
|
|
const pendingItems = await appDb
|
|
|
|
|
.select()
|
|
|
|
|
.from(importJobItems)
|
|
|
|
|
.where(eq(importJobItems.status, "pending"));
|
|
|
|
|
|
|
|
|
|
// --- Concurrent item processing ---
|
|
|
|
|
let idx = 0;
|
|
|
|
|
const worker = async () => {
|
|
|
|
|
while (true) {
|
|
|
|
|
const item = pendingItems[idx++];
|
|
|
|
|
if (!item) break;
|
|
|
|
|
|
|
|
|
|
await appDb
|
|
|
|
|
.update(importJobItems)
|
|
|
|
|
.set({ status: "processing" })
|
|
|
|
|
.where(eq(importJobItems.id, item.id));
|
|
|
|
|
|
|
|
|
|
try {
|
|
|
|
|
const { immichId } = await processor.processItem(job, item, processorContext);
|
|
|
|
|
await appDb
|
|
|
|
|
.update(importJobItems)
|
|
|
|
|
.set({ status: "uploaded", immichId })
|
|
|
|
|
.where(eq(importJobItems.id, item.id));
|
|
|
|
|
await appDb
|
|
|
|
|
.update(importJobs)
|
|
|
|
|
.set({ uploadedCount: job.uploadedCount + 1 })
|
|
|
|
|
.where(eq(importJobs.id, jobId));
|
|
|
|
|
} catch (err: any) {
|
|
|
|
|
await appDb
|
|
|
|
|
.update(importJobItems)
|
|
|
|
|
.set({ status: "failed", error: err?.message ?? "Unknown error" })
|
|
|
|
|
.where(eq(importJobItems.id, item.id));
|
|
|
|
|
await appDb
|
|
|
|
|
.update(importJobs)
|
|
|
|
|
.set({ failedCount: job.failedCount + 1 })
|
|
|
|
|
.where(eq(importJobs.id, jobId));
|
|
|
|
|
}
|
|
|
|
|
}
|
|
|
|
|
};
|
|
|
|
|
|
|
|
|
|
const concurrency = Math.min(CONCURRENCY, pendingItems.length);
|
|
|
|
|
await Promise.all(Array.from({ length: concurrency }, worker));
|
|
|
|
|
|
|
|
|
|
await appDb
|
|
|
|
|
.update(importJobs)
|
|
|
|
|
.set({ status: "completed" })
|
|
|
|
|
.where(eq(importJobs.id, jobId));
|
|
|
|
|
} catch (err: any) {
|
|
|
|
|
console.error(`Import job ${jobId} failed fatally:`, err);
|
|
|
|
|
await appDb
|
|
|
|
|
.update(importJobs)
|
|
|
|
|
.set({ status: "failed" })
|
|
|
|
|
.where(eq(importJobs.id, jobId));
|
|
|
|
|
} finally {
|
|
|
|
|
activeJobs.delete(jobId);
|
|
|
|
|
}
|
|
|
|
|
}
|
|
|
|
|
```
|
|
|
|
|
|
|
|
|
|
**Step 2: Commit**
|
|
|
|
|
|
|
|
|
|
```bash
|
|
|
|
|
git add src/workers/import/runner.ts
|
|
|
|
|
git commit -m "feat: add import job worker runner with concurrency and db tracking"
|
|
|
|
|
```
|
|
|
|
|
|
|
|
|
|
---
|
|
|
|
|
|
|
|
|
|
### Task 6: Update instrumentation.ts for startup resume
|
|
|
|
|
|
|
|
|
|
**Files:**
|
|
|
|
|
- Modify: `instrumentation.ts`
|
|
|
|
|
|
|
|
|
|
**Step 1: Add resume call**
|
|
|
|
|
|
|
|
|
|
```typescript
|
|
|
|
|
// instrumentation.ts
|
|
|
|
|
export async function register() {
|
|
|
|
|
if (process.env.NEXT_RUNTIME === "nodejs") {
|
|
|
|
|
const { runMigrations } = await import("./src/db");
|
|
|
|
|
await runMigrations();
|
|
|
|
|
|
|
|
|
|
const { resumePendingJobs } = await import("./src/workers/import/runner");
|
|
|
|
|
await resumePendingJobs();
|
|
|
|
|
}
|
|
|
|
|
}
|
|
|
|
|
```
|
|
|
|
|
|
|
|
|
|
`resumePendingJobs` only resets `processing → pending`; it doesn't start workers (no user headers available at boot). Workers restart on next poll.
|
|
|
|
|
|
|
|
|
|
**Step 2: Commit**
|
|
|
|
|
|
|
|
|
|
```bash
|
|
|
|
|
git add instrumentation.ts
|
|
|
|
|
git commit -m "feat: reset stuck import jobs to pending on startup"
|
|
|
|
|
```
|
|
|
|
|
|
|
|
|
|
---
|
|
|
|
|
|
|
|
|
|
### Task 7: Job creation API endpoint
|
|
|
|
|
|
|
|
|
|
**Files:**
|
|
|
|
|
- Create: `src/pages/api/import-jobs/index.ts`
|
|
|
|
|
|
|
|
|
|
**Step 1: Create the endpoint**
|
|
|
|
|
|
|
|
|
|
```typescript
|
|
|
|
|
// src/pages/api/import-jobs/index.ts
|
|
|
|
|
import type { NextApiRequest, NextApiResponse } from "next";
|
|
|
|
|
import { appDb } from "@/db";
|
|
|
|
|
import { importJobs, importJobItems } from "@/db/schema";
|
|
|
|
|
import { getCurrentUser } from "@/handlers/serverUtils/user.utils";
|
|
|
|
|
import { getUserHeaders } from "@/helpers/user.helper";
|
|
|
|
|
import { runImportJob } from "@/workers/import/runner";
|
|
|
|
|
|
|
|
|
|
export default async function handler(req: NextApiRequest, res: NextApiResponse) {
|
|
|
|
|
if (req.method !== "POST") {
|
|
|
|
|
res.setHeader("Allow", "POST");
|
|
|
|
|
return res.status(405).json({ error: "Method Not Allowed" });
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
const currentUser = await getCurrentUser(req);
|
|
|
|
|
if (!currentUser) return res.status(401).json({ error: "Unauthorized" });
|
|
|
|
|
|
|
|
|
|
const { platform, url, urlConfig, importData, assets } = req.body;
|
|
|
|
|
|
|
|
|
|
if (!platform || !url || !Array.isArray(assets) || assets.length === 0) {
|
|
|
|
|
return res.status(400).json({ error: "platform, url, and assets are required" });
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
// Create job row
|
|
|
|
|
const [job] = await appDb
|
|
|
|
|
.insert(importJobs)
|
|
|
|
|
.values({
|
|
|
|
|
userId: currentUser.id,
|
|
|
|
|
platform,
|
|
|
|
|
url,
|
|
|
|
|
urlConfig: JSON.stringify(urlConfig ?? {}),
|
|
|
|
|
importData: JSON.stringify(importData ?? {}),
|
|
|
|
|
totalCount: assets.length,
|
|
|
|
|
})
|
|
|
|
|
.returning();
|
|
|
|
|
|
|
|
|
|
// Create one item row per asset
|
|
|
|
|
await appDb.insert(importJobItems).values(
|
|
|
|
|
assets.map((asset: { id: string; [key: string]: unknown }) => ({
|
|
|
|
|
jobId: job.id,
|
|
|
|
|
assetId: asset.id,
|
|
|
|
|
itemData: JSON.stringify(asset),
|
|
|
|
|
}))
|
|
|
|
|
);
|
|
|
|
|
|
|
|
|
|
// Fire-and-forget: start the worker without awaiting
|
|
|
|
|
const headers = getUserHeaders(currentUser) as Record<string, string>;
|
|
|
|
|
runImportJob(job.id, headers).catch((err) =>
|
|
|
|
|
console.error("Unhandled error in runImportJob", err)
|
|
|
|
|
);
|
|
|
|
|
|
|
|
|
|
return res.status(201).json({ jobId: job.id });
|
|
|
|
|
}
|
|
|
|
|
```
|
|
|
|
|
|
|
|
|
|
**Step 2: Commit**
|
|
|
|
|
|
|
|
|
|
```bash
|
|
|
|
|
git add src/pages/api/import-jobs/index.ts
|
|
|
|
|
git commit -m "feat: add POST /api/import-jobs endpoint to create and enqueue import jobs"
|
|
|
|
|
```
|
|
|
|
|
|
|
|
|
|
---
|
|
|
|
|
|
|
|
|
|
### Task 8: Job status polling endpoint
|
|
|
|
|
|
|
|
|
|
**Files:**
|
|
|
|
|
- Create: `src/pages/api/import-jobs/[jobId].ts`
|
|
|
|
|
|
|
|
|
|
**Step 1: Create the endpoint**
|
|
|
|
|
|
|
|
|
|
```typescript
|
|
|
|
|
// src/pages/api/import-jobs/[jobId].ts
|
|
|
|
|
import type { NextApiRequest, NextApiResponse } from "next";
|
|
|
|
|
import { eq } from "drizzle-orm";
|
|
|
|
|
import { appDb } from "@/db";
|
|
|
|
|
import { importJobs, importJobItems } from "@/db/schema";
|
|
|
|
|
import { getCurrentUser } from "@/handlers/serverUtils/user.utils";
|
|
|
|
|
import { getUserHeaders } from "@/helpers/user.helper";
|
|
|
|
|
import { isJobActive, runImportJob } from "@/workers/import/runner";
|
|
|
|
|
|
|
|
|
|
export default async function handler(req: NextApiRequest, res: NextApiResponse) {
|
|
|
|
|
if (req.method !== "GET") {
|
|
|
|
|
res.setHeader("Allow", "GET");
|
|
|
|
|
return res.status(405).json({ error: "Method Not Allowed" });
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
const currentUser = await getCurrentUser(req);
|
|
|
|
|
if (!currentUser) return res.status(401).json({ error: "Unauthorized" });
|
|
|
|
|
|
|
|
|
|
const { jobId } = req.query as { jobId: string };
|
|
|
|
|
|
|
|
|
|
const [job] = await appDb.select().from(importJobs).where(eq(importJobs.id, jobId));
|
|
|
|
|
if (!job || job.userId !== currentUser.id) {
|
|
|
|
|
return res.status(404).json({ error: "Job not found" });
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
// Auto-resume: if job is pending but has no active worker (e.g. after server restart)
|
|
|
|
|
if (job.status === "pending" && !isJobActive(jobId)) {
|
|
|
|
|
const headers = getUserHeaders(currentUser) as Record<string, string>;
|
|
|
|
|
runImportJob(jobId, headers).catch((err) =>
|
|
|
|
|
console.error("Unhandled error in runImportJob (resume)", err)
|
|
|
|
|
);
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
const items = await appDb
|
|
|
|
|
.select()
|
|
|
|
|
.from(importJobItems)
|
|
|
|
|
.where(eq(importJobItems.jobId, jobId));
|
|
|
|
|
|
|
|
|
|
return res.status(200).json({ job, items });
|
|
|
|
|
}
|
|
|
|
|
```
|
|
|
|
|
|
|
|
|
|
**Step 2: Commit**
|
|
|
|
|
|
|
|
|
|
```bash
|
|
|
|
|
git add src/pages/api/import-jobs/[jobId].ts
|
|
|
|
|
git commit -m "feat: add GET /api/import-jobs/[jobId] polling endpoint with auto-resume"
|
|
|
|
|
```
|
|
|
|
|
|
|
|
|
|
---
|
|
|
|
|
|
|
|
|
|
### Task 9: Route constants and client handler
|
|
|
|
|
|
|
|
|
|
**Files:**
|
|
|
|
|
- Modify: `src/config/routes.ts`
|
|
|
|
|
- Create: `src/handlers/api/import-jobs.handler.ts`
|
|
|
|
|
|
|
|
|
|
**Step 1: Add routes**
|
|
|
|
|
|
|
|
|
|
In `src/config/routes.ts`, append:
|
|
|
|
|
```typescript
|
|
|
|
|
// Import Jobs (queue)
|
|
|
|
|
export const IMPORT_JOBS_PATH = BASE_API_ENDPOINT + "/import-jobs";
|
|
|
|
|
export const IMPORT_JOB_PATH = (jobId: string) => BASE_API_ENDPOINT + "/import-jobs/" + jobId;
|
|
|
|
|
```
|
|
|
|
|
|
|
|
|
|
**Step 2: Create handler**
|
|
|
|
|
|
|
|
|
|
```typescript
|
|
|
|
|
// src/handlers/api/import-jobs.handler.ts
|
|
|
|
|
import { IMPORT_JOBS_PATH, IMPORT_JOB_PATH } from "@/config/routes";
|
|
|
|
|
import API from "@/lib/api";
|
|
|
|
|
|
|
|
|
|
export interface CreateImportJobParams {
|
|
|
|
|
platform: string;
|
|
|
|
|
url: string;
|
|
|
|
|
urlConfig: Record<string, unknown>;
|
|
|
|
|
importData: Record<string, unknown>;
|
|
|
|
|
assets: { id: string; [key: string]: unknown }[];
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
export const createImportJob = async (params: CreateImportJobParams): Promise<{ jobId: string }> => {
|
|
|
|
|
return API.post(IMPORT_JOBS_PATH, params);
|
|
|
|
|
};
|
|
|
|
|
|
|
|
|
|
export const getImportJob = async (jobId: string) => {
|
|
|
|
|
return API.get(IMPORT_JOB_PATH(jobId));
|
|
|
|
|
};
|
|
|
|
|
```
|
|
|
|
|
|
|
|
|
|
**Step 3: Commit**
|
|
|
|
|
|
|
|
|
|
```bash
|
|
|
|
|
git add src/config/routes.ts src/handlers/api/import-jobs.handler.ts
|
|
|
|
|
git commit -m "feat: add import-jobs route constants and client handler"
|
|
|
|
|
```
|
|
|
|
|
|
|
|
|
|
---
|
|
|
|
|
|
|
|
|
|
### Task 10: Update frontend to use job queue + polling
|
|
|
|
|
|
|
|
|
|
**Files:**
|
|
|
|
|
- Modify: `src/pages/assets/import-shared.tsx`
|
|
|
|
|
|
|
|
|
|
**Step 1: Add state for active job**
|
|
|
|
|
|
|
|
|
|
In the component state block, add:
|
|
|
|
|
```typescript
|
|
|
|
|
const [activeJobId, setActiveJobId] = useState<string | null>(null);
|
|
|
|
|
const [jobProgress, setJobProgress] = useState<{ uploaded: number; skipped: number; failed: number; total: number } | null>(null);
|
|
|
|
|
```
|
|
|
|
|
|
|
|
|
|
**Step 2: Replace `handleImportAll` body**
|
|
|
|
|
|
|
|
|
|
Replace the `importSharedUploadAll(...)` call block with:
|
|
|
|
|
|
|
|
|
|
```typescript
|
|
|
|
|
const job = await createImportJob({
|
|
|
|
|
platform: "immich",
|
|
|
|
|
url: sharedData.origin,
|
|
|
|
|
urlConfig: { key: sharedData.key },
|
|
|
|
|
importData: { albumOptions },
|
|
|
|
|
assets: uploadableAssets.map((asset) => ({
|
|
|
|
|
id: asset.id,
|
|
|
|
|
originalFileName: asset.originalFileName,
|
|
|
|
|
type: asset.type,
|
|
|
|
|
fileCreatedAt: asset.fileCreatedAt ?? null,
|
|
|
|
|
localDateTime: asset.localDateTime ?? null,
|
|
|
|
|
duration: asset.duration ?? null,
|
|
|
|
|
isFavorite: asset.isFavorite ?? false,
|
|
|
|
|
isArchived: asset.isArchived ?? false,
|
|
|
|
|
})),
|
|
|
|
|
});
|
|
|
|
|
setActiveJobId(job.jobId);
|
|
|
|
|
setJobProgress({ uploaded: 0, skipped: 0, failed: 0, total: uploadableAssets.length });
|
|
|
|
|
// importAllLoading stays true while polling
|
|
|
|
|
```
|
|
|
|
|
|
|
|
|
|
Remove the old result-parsing and banner-setting code from `handleImportAll` — that moves to the polling effect below.
|
|
|
|
|
|
|
|
|
|
**Step 3: Add polling effect**
|
|
|
|
|
|
|
|
|
|
```typescript
|
|
|
|
|
useEffect(() => {
|
|
|
|
|
if (!activeJobId) return;
|
|
|
|
|
|
|
|
|
|
const interval = setInterval(async () => {
|
|
|
|
|
try {
|
|
|
|
|
const { job, items } = await getImportJob(activeJobId);
|
|
|
|
|
setJobProgress({
|
|
|
|
|
uploaded: job.uploadedCount,
|
|
|
|
|
skipped: job.skippedCount,
|
|
|
|
|
failed: job.failedCount,
|
|
|
|
|
total: job.totalCount,
|
|
|
|
|
});
|
|
|
|
|
|
|
|
|
|
if (job.status === "completed" || job.status === "failed") {
|
|
|
|
|
clearInterval(interval);
|
|
|
|
|
setActiveJobId(null);
|
|
|
|
|
setImportAllLoading(false);
|
|
|
|
|
|
|
|
|
|
const importData = JSON.parse(job.importData);
|
|
|
|
|
const failedCount = job.failedCount;
|
|
|
|
|
const skippedCount = job.skippedCount;
|
|
|
|
|
const uploadedCount = job.uploadedCount;
|
|
|
|
|
let bannerType: "success" | "error" = failedCount ? "error" : "success";
|
|
|
|
|
let message: string;
|
|
|
|
|
|
|
|
|
|
if (!uploadedCount && skippedCount && !failedCount) {
|
|
|
|
|
message = `All ${skippedCount} assets already exist on Immich.`;
|
|
|
|
|
} else if (failedCount) {
|
|
|
|
|
const skippedSuffix = skippedCount ? `, ${skippedCount} skipped` : "";
|
|
|
|
|
message = `Imported ${uploadedCount} assets, ${failedCount} failed${skippedSuffix}.`;
|
|
|
|
|
} else if (skippedCount) {
|
|
|
|
|
message = `Imported ${uploadedCount} assets, skipped ${skippedCount} already on Immich.`;
|
|
|
|
|
} else {
|
|
|
|
|
message = `Imported ${uploadedCount} assets successfully.`;
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
if (importData.albumId && uploadedCount > 0) {
|
|
|
|
|
const albumName = options.albumName || "selected album";
|
|
|
|
|
message = `${message} Added to album "${albumName}".`;
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
setUploadBanner({ type: bannerType, message });
|
|
|
|
|
}
|
|
|
|
|
} catch (err) {
|
|
|
|
|
console.error("Polling error", err);
|
|
|
|
|
}
|
|
|
|
|
}, 2000);
|
|
|
|
|
|
|
|
|
|
return () => clearInterval(interval);
|
|
|
|
|
}, [activeJobId]);
|
|
|
|
|
```
|
|
|
|
|
|
|
|
|
|
**Step 4: Update the progress UI**
|
|
|
|
|
|
|
|
|
|
Replace the spinner/loading button text with a live counter when `jobProgress` is set:
|
|
|
|
|
|
|
|
|
|
```tsx
|
|
|
|
|
{importAllLoading && jobProgress && (
|
|
|
|
|
<p className="text-sm text-muted-foreground text-center">
|
|
|
|
|
{jobProgress.uploaded + jobProgress.skipped} / {jobProgress.total} processed
|
|
|
|
|
{jobProgress.failed > 0 && ` · ${jobProgress.failed} failed`}
|
|
|
|
|
</p>
|
|
|
|
|
)}
|
|
|
|
|
```
|
|
|
|
|
|
|
|
|
|
**Step 5: Add imports at the top of the file**
|
|
|
|
|
|
|
|
|
|
```typescript
|
|
|
|
|
import { createImportJob, getImportJob } from "@/handlers/api/import-jobs.handler";
|
|
|
|
|
```
|
|
|
|
|
|
|
|
|
|
**Step 6: Commit**
|
|
|
|
|
|
|
|
|
|
```bash
|
|
|
|
|
git add src/pages/assets/import-shared.tsx
|
|
|
|
|
git commit -m "feat: migrate import-shared to job queue with live progress polling"
|
|
|
|
|
```
|
|
|
|
|
|
|
|
|
|
---
|
|
|
|
|
|
|
|
|
|
## Done
|
|
|
|
|
|
|
|
|
|
At this point:
|
|
|
|
|
- Imports are non-blocking: the page can be navigated away and the import continues server-side
|
|
|
|
|
- Progress is visible: `X / Y processed` updates every 2s
|
|
|
|
|
- Server restarts are handled: `instrumentation.ts` resets stuck jobs; next poll auto-resumes
|
|
|
|
|
- New platforms (Nextcloud, PhotoPrism) only need a new processor file + `registerProcessor` call in `processors/index.ts`
|
|
|
|
|
- The old `/api/import-shared/upload-all` endpoint is untouched and can be removed in a follow-up
|
Varun