diff --git a/AGENTS.md b/AGENTS.md index 0c8a236..06e4682 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -128,8 +128,8 @@ platform support that the release configuration does not provide. ## Gotchas -- `~/.rawback/{config.yml,credentials.json,upload-state.json,cameras.json}` and - `~/.rawback/logs/` are shared with the Desktop app at runtime. Changing a +- `~/.rawback/{config.yml,credentials.json,upload-state.json,upload-ledger.jsonl,cameras.json}` + and `~/.rawback/logs/` are shared with the Desktop app at runtime. Changing a file's shape here breaks Desktop, and the contract is owned by `@rawback/sdk`, not by this repo. `rawback logs purge` clears Desktop's log too, by design. - `@rawback/sdk` and `@rawback/ccapi-js` are pinned to exact versions. Bumping diff --git a/README.md b/README.md index dfaeeee..8c7683c 100644 --- a/README.md +++ b/README.md @@ -473,13 +473,14 @@ anything is written. Rawback stores local state under `~/.rawback/`: -| File | Purpose | -| ------------------- | -------------------------------------------------------------- | -| `credentials.json` | Access and refresh tokens per environment, from `rawback auth` | -| `config.yml` | Environments, optional hosts, metadata workers, and SFTP | -| `upload-state.json` | Shared upload queue, history, and trusted host keys | -| `cameras.json` | Saved Canon cameras, shared with Rawback Desktop | -| `logs/` | Rolling JSON diagnostics, shared with Rawback Desktop | +| File | Purpose | +| --------------------- | -------------------------------------------------------------- | +| `credentials.json` | Access and refresh tokens per environment, from `rawback auth` | +| `config.yml` | Environments, optional hosts, metadata workers, and SFTP | +| `upload-state.json` | Shared upload queue, history, and trusted host keys | +| `upload-ledger.jsonl` | Files already in your library, shared with Rawback Desktop | +| `cameras.json` | Saved Canon cameras, shared with Rawback Desktop | +| `logs/` | Rolling JSON diagnostics, shared with Rawback Desktop | `cameras.json` is shared with the Rawback desktop app, so both can reach the same camera without pairing twice. It holds a camera password only when you pass diff --git a/bun.lock b/bun.lock index a235db5..d7ae328 100644 --- a/bun.lock +++ b/bun.lock @@ -7,7 +7,7 @@ "dependencies": { "@inquirer/prompts": "^8.7.0", "@rawback/ccapi-js": "1.1.3", - "@rawback/sdk": "0.3.5", + "@rawback/sdk": "0.3.6", "ink": "^7.1.1", "react": "^19.2.8", "react-devtools-core": "^7.0.1", @@ -149,7 +149,7 @@ "@rawback/ccapi-js": ["@rawback/ccapi-js@1.1.3", "", {}, "sha512-+ixw0xQw+YDZvHwUeYnum89/ZxxVPanJKxeSEkV9jXMkXhXtOarUz3a0ZrTRz7ia3H9FWJJpsu8rW4Hdx+I6VQ=="], - "@rawback/sdk": ["@rawback/sdk@0.3.5", "", { "dependencies": { "@graphql-typed-document-node/core": "3.2.0", "blurhash": "2.0.5", "exiftool-vendored": "37.0.0", "graphql": "17.0.2", "pino": "^10.3.1", "pino-roll": "^4.0.0", "proper-lockfile": "4.1.2", "ssh2": "1.17.0", "yaml": "2.9.0", "zod": "4.4.3" } }, "sha512-y5llQyWT6keIA1Dm9BCziYa0oZVyibYC8k2/v6aSm36c45vT1VjZJP/tUSmHNSwV011IWkENKGBfIwCwX/tVaw=="], + "@rawback/sdk": ["@rawback/sdk@0.3.6", "", { "dependencies": { "@graphql-typed-document-node/core": "3.2.0", "blurhash": "2.0.5", "exiftool-vendored": "37.0.0", "graphql": "17.0.2", "pino": "^10.3.1", "pino-roll": "^4.0.0", "proper-lockfile": "4.1.2", "ssh2": "1.17.0", "yaml": "2.9.0", "zod": "4.4.3" } }, "sha512-Z5cFpxgahDVV6CMWvX1Fivw+PUMyB9JmgspDPphOUnN+LUA9chQ6uyyTTksjj02lw0vih/qSl96Vo0TLwKTCEw=="], "@types/bun": ["@types/bun@1.4.2", "", { "dependencies": { "bun-types": "1.4.2" } }, "sha512-GimotNn7+ZV0uVArItBbriZsR1oNf0+WTzPkdcFrzShI7k2norL0uzEaJT8T33dWr7O/c9ZDuAFQrctKCi72oQ=="], diff --git a/docs/commands.md b/docs/commands.md index 42b2c14..c719912 100644 --- a/docs/commands.md +++ b/docs/commands.md @@ -670,6 +670,7 @@ rawback photos upload --path [options] | `--path ` | Required image/RAW file or recursively scanned directory | — | | `--concurrency ` | Parallel SFTP transfers; integer from 1 through 16 | `4` | | `--dry-run` | Report pending/skipped files, bytes, and estimated duration | `false` | +| `--recheck` | Ignore local upload history; ask the server about each file | `false` | Upload setup and safety behavior are covered in [Configuration and uploads](configuration.md). Unsupported files are skipped @@ -679,6 +680,13 @@ EXIF capture time; duplicate-check failures do not block the SFTP upload. Local EXIF workers are controlled separately by `metadata.concurrency` in the shared configuration file. +After the transfer, the server is asked which uploaded photos it now holds; the +summary reports them as Confirmed, and any it does not have as Unconfirmed (they +upload again next time). Acknowledged files are remembered in +`~/.rawback/upload-ledger.jsonl`, shared with Rawback Desktop, so the next run +skips them before reading metadata; see +[Upload acknowledgements](configuration.md#upload-acknowledgements). + ## `rawback spots` Browses public photography Spots: places where Rawback photographers published diff --git a/docs/configuration.md b/docs/configuration.md index ec06bc9..3680f87 100644 --- a/docs/configuration.md +++ b/docs/configuration.md @@ -360,7 +360,7 @@ rawback photos upload --path /path/to/photos --dry-run The dry run authenticates, validates configuration and account quota, scans the input, and checks exact remote upload identities. It does not open an SFTP connection or -modify upload state. Its time estimate uses matching local upload +modify upload state or the upload ledger. Its time estimate uses matching local upload history when available and otherwise uses a clearly labeled 10 Mbps fallback. Start the upload after reviewing the totals: @@ -370,12 +370,12 @@ rawback photos upload --path /path/to/photos --concurrency 4 ``` Transfer concurrency defaults to `4` and accepts integers from `1` through `16`. All +workers share one SFTP connection. While an upload is running in an interactive terminal, the CLI shows aggregate bytes and files, transfer speed, ETA, and up to four active filenames. When stdout is redirected, it emits line-oriented file status and a final summary instead of terminal animation. -workers share one SFTP connection. Before connecting, the CLI verifies: @@ -404,14 +404,40 @@ Supported image extensions are `.jpg`, `.jpeg`, `.png`, `.webp`, `.gif`, `.tif`, and `.x3f`. Matching is case-insensitive. Directory scanning skips symbolic links and unrelated file types, but fails if it finds no supported files. +## Upload acknowledgements + +A successful SFTP transfer does not say whether Rawback stored the photo, found +it to be a duplicate, or failed to record it. After the transfer, the CLI asks +the server which uploaded photos it now holds (matching filename and capture +time), and the final summary reports them as **Confirmed**. A photo the server +does not have is reported as **Unconfirmed** and is uploaded again next time. + +Every acknowledged file is remembered in `~/.rawback/upload-ledger.jsonl`, which +Rawback Desktop reads and writes too, for the signed-in account on the current +environment's API host. A later run — from either app — skips a remembered +file before reading its metadata or asking the server: + +- A file is matched by canonical path, size, and modification time; a photo + confirmed at another path (a moved folder, a second copy) is matched by + filename and capture time. +- Photos the server confirmed, and photos without a capture time (which the + server can neither confirm nor deduplicate), are skipped outright. +- A photo whose confirmation request failed is checked with the server again + instead, without re-reading its metadata. + +Pass `--recheck` to ignore the ledger and the resume state and ask the server +about every file; what it learns is still recorded. + ## Resume state and interruption -Completed uploads are recorded in the portable `~/.rawback/upload-state.json` +Uploads in progress are recorded in the portable `~/.rawback/upload-state.json` shared with Rawback Desktop. On first use, the CLI imports its legacy -`upload-progress.sqlite` records and retains that database for rollback. A later -run skips a completed file only when its canonical path, size, and modification -time still match. Incomplete files restart from the beginning because the SFTP -transport does not append partial content. +`upload-progress.sqlite` records and retains that database for rollback. A +completed file recorded there before the upload ledger existed is skipped only +when its canonical path, size, and modification time still match and it has no +capture time; other photos are checked with the server once and remembered in +the ledger from then on. Incomplete files restart from the beginning because +the SFTP transport does not append partial content. On the first `Ctrl-C`, the CLI stops scheduling new files and lets active uploads finish. Press `Ctrl-C` again to disconnect immediately. Re-run the same command diff --git a/src/cli.ts b/src/cli.ts index 22216dc..30f2cb6 100644 --- a/src/cli.ts +++ b/src/cli.ts @@ -675,6 +675,11 @@ export function createProgram(version: string, output = new CommandOutput()): Ar describe: 'show upload count, size, and estimated time without uploading', type: 'boolean', }) + .option('recheck', { + default: false, + describe: 'ask the server about every file; ignore local upload history', + type: 'boolean', + }) .check((args) => { if ( !Number.isInteger(args.concurrency) || @@ -693,6 +698,7 @@ export function createProgram(version: string, output = new CommandOutput()): Ar concurrency: args.concurrency, dryRun: args.dryRun, path: args.path, + recheck: args.recheck, }), ) }, diff --git a/src/features/upload/view.ts b/src/features/upload/view.ts index 26c0f98..fdbf28a 100644 --- a/src/features/upload/view.ts +++ b/src/features/upload/view.ts @@ -2,7 +2,7 @@ import { formatBytes, formatDuration } from '../../ui/format.ts' import { type UiDocument } from '../../ui/model.ts' export function uploadDryRunDocument(input: { - completed: number + alreadyUploaded: number estimatedSeconds: number existingRemote: number files: number @@ -18,7 +18,7 @@ export function uploadDryRunDocument(input: { fields: [ { label: 'Files', value: input.files }, { label: 'Total size', value: formatBytes(input.totalBytes) }, - { label: 'Completed', value: input.completed }, + { label: 'Already uploaded', value: input.alreadyUploaded }, { label: 'Remote', value: input.existingRemote }, { label: 'Estimated time', value: formatDuration(input.estimatedSeconds) }, { @@ -34,8 +34,12 @@ export function uploadDryRunDocument(input: { export function uploadSummaryDocument(input: { cancelled: boolean + /** Uploaded photos the server confirmed holding. */ + confirmed: number failedFiles: number totalBytes: number + /** Uploaded photos the server was asked about and did not have. */ + unconfirmed: number }): UiDocument { const tone = input.cancelled || input.failedFiles > 0 ? 'warning' : 'success' return { @@ -54,6 +58,8 @@ export function uploadSummaryDocument(input: { type: 'fields', fields: [ { label: 'Transferred', value: formatBytes(input.totalBytes) }, + { label: 'Confirmed', value: input.confirmed }, + ...(input.unconfirmed > 0 ? [{ label: 'Unconfirmed', value: input.unconfirmed }] : []), { label: 'Failed', value: input.failedFiles }, ], }, diff --git a/src/upload-ledger.ts b/src/upload-ledger.ts new file mode 100644 index 0000000..31c0320 --- /dev/null +++ b/src/upload-ledger.ts @@ -0,0 +1,234 @@ +import { dirname, join } from 'node:path' + +import { + type GraphqlCompatibilityClient, + matchUploadIdentities, + type UploadAck, + UploadLedger, + type UploadLedgerChange, + type UploadLedgerEntry, + type UploadLedgerSource, + uploadLedgerScope, +} from '@rawback/sdk' + +export const UPLOAD_LEDGER_FILENAME = 'upload-ledger.jsonl' + +/** + * The ledger lives beside the upload state — `~/.rawback/upload-ledger.jsonl`, + * the file Desktop reads too — so a test that points the state elsewhere never + * touches the real one. + */ +export function ledgerPathFor(statePath: string): string { + return join(dirname(statePath), UPLOAD_LEDGER_FILENAME) +} + +/** The fields of an upload candidate the ledger reads and writes. */ +export interface LedgerFile { + basename: string + canonicalPath: string + capturedAt?: string + clientKey?: string + mtimeMs: number + size: number +} + +export interface LedgerReview { + /** Files the ledger settles: skipped without reading metadata or asking the server. */ + skipped: File[] + /** + * Files sent before whose confirmation never came back. Their capture time + * comes from the ledger, so they skip the metadata read but still go to the + * server's check. + */ + recheck: File[] + /** Files the ledger knows nothing about. */ + unknown: File[] +} + +export interface Confirmation { + confirmed: number + unconfirmed: number +} + +/** + * Whether an entry is enough to skip a file without asking the server — the + * same rule `UploadManager` applies. A `server` acknowledgement is; a + * `transfer` one only when the photo has no capture time, because the server + * can neither confirm nor deduplicate such a photo. A `transfer` + * acknowledgement with a capture time means the confirmation failed, and the + * gateway answers OK even when it failed to record a photo, so it is checked + * again. + */ +export function isSettled(entry: Pick): boolean { + return entry.ack === 'server' || !entry.capturedAt +} + +/** + * This run's view of the upload ledger shared with Desktop: which files are + * already in the account's library, and the acknowledgements this run adds. + * + * Every failure here is a warning, never an error: the ledger only saves work, + * and the gateway and the server's identity check still stop duplicates + * without it. + */ +export class UploadAcknowledgements { + private pending: UploadLedgerChange[] = [] + + private constructor( + private readonly ledger: UploadLedger | undefined, + private readonly scope: string | undefined, + private readonly warn: (message: string) => void, + ) {} + + static async open(options: { + path: string + apiHost: string | undefined + accountId: number | string + warn: (message: string) => void + }): Promise { + let scope: string + try { + scope = uploadLedgerScope({ apiHost: options.apiHost ?? '', accountId: options.accountId }) + } catch { + options.warn('The API host is not an http(s) URL; uploads will not be remembered locally.') + return new UploadAcknowledgements(undefined, undefined, options.warn) + } + const ledger = new UploadLedger({ path: options.path }) + try { + await ledger.load() + } catch { + options.warn( + `The upload ledger at ${options.path} could not be read; every file will be checked.`, + ) + return new UploadAcknowledgements(undefined, undefined, options.warn) + } + return new UploadAcknowledgements(ledger, scope, options.warn) + } + + /** Sorts files by what the ledger knows about them, before any metadata is read. */ + review(files: File[]): LedgerReview { + const review: LedgerReview = { skipped: [], recheck: [], unknown: [] } + for (const file of files) { + const entry = this.ledger && this.scope ? this.ledger.findByFile(this.scope, file) : undefined + if (!entry) review.unknown.push(file) + else if (isSettled(entry)) review.skipped.push(file) + else { + review.recheck.push({ + ...file, + capturedAt: entry.capturedAt!, + clientKey: crypto.randomUUID(), + }) + } + } + return review + } + + /** + * Whether the server has confirmed this photo at another path — a moved + * folder or a second copy. A match is recorded under the new path so the + * next run takes the fast path for it. + */ + skipByIdentity(file: LedgerFile): boolean { + if (!this.ledger || !this.scope || !file.capturedAt) return false + const entry = this.ledger.findByIdentity(this.scope, { + name: file.basename, + capturedAt: file.capturedAt, + }) + if (!entry || !isSettled(entry)) return false + this.record(file, entry.ack, entry.source, entry.imageId) + return true + } + + record(file: LedgerFile, ack: UploadAck, source: UploadLedgerSource, imageId?: number): void { + if (!this.scope) return + this.pending.push({ + kind: 'record', + record: { + scope: this.scope, + path: file.canonicalPath, + size: file.size, + mtimeMs: file.mtimeMs, + name: file.basename, + ...(file.capturedAt ? { capturedAt: file.capturedAt } : {}), + ...(imageId === undefined ? {} : { imageId }), + ack, + source, + }, + }) + } + + forget(file: LedgerFile): void { + if (!this.scope) return + this.pending.push({ kind: 'forget', scope: this.scope, path: file.canonicalPath }) + } + + /** Writes what this run has learned. A failed write is a warning. */ + async flush(): Promise { + const changes = this.pending.splice(0) + if (!this.ledger || changes.length === 0) return + try { + await this.ledger.apply(changes) + } catch { + this.warn('Uploaded files could not be recorded in the local upload ledger.') + } + } + + /** + * Acknowledges files the SFTP server accepted: a `transfer` acknowledgement, + * then the server's own answer for every photo with a capture time. The + * gateway answers a close with OK for a new photo, a duplicate and one it + * failed to record alike, so only that answer tells them apart. A photo the + * server does not have is forgotten so the next run uploads it again; a + * failed request keeps the transfer acknowledgement, which the next run + * checks again. + * + * Everything is written in one append once the answer is in, so a confirmed + * photo costs one ledger line. Nothing is lost by waiting: until then the + * upload state's resume record covers a photo, and a transfer + * acknowledgement would not have let a photo with a capture time be skipped + * anyway. + */ + async acknowledge( + graphql: GraphqlCompatibilityClient, + uploaded: LedgerFile[], + ): Promise { + for (const file of uploaded) this.record(file, 'transfer', 'upload') + const candidates = uploaded.filter( + (file): file is LedgerFile & { capturedAt: string; clientKey: string } => + Boolean(file.capturedAt && file.clientKey), + ) + const confirmation: Confirmation = { confirmed: 0, unconfirmed: 0 } + if (candidates.length === 0) { + await this.flush() + return confirmation + } + + let matches: Map + try { + matches = await matchUploadIdentities( + graphql, + candidates.map((file) => ({ + clientKey: file.clientKey, + originalFilename: file.basename, + capturedAt: file.capturedAt, + })), + ) + } catch { + this.warn('Rawback could not confirm the uploaded photos; they will be checked next time.') + await this.flush() + return confirmation + } + for (const file of candidates) { + const imageId = matches.get(file.clientKey) + if (imageId === undefined) { + this.forget(file) + confirmation.unconfirmed += 1 + } else { + this.record(file, 'server', 'upload', imageId) + confirmation.confirmed += 1 + } + } + await this.flush() + return confirmation + } +} diff --git a/src/upload.ts b/src/upload.ts index ca49585..c49139a 100644 --- a/src/upload.ts +++ b/src/upload.ts @@ -2,9 +2,9 @@ import { lstat, readdir, realpath } from 'node:fs/promises' import { basename, join, resolve } from 'node:path' import { - ExistingUploadIdentitiesDocument, UploadPreflightDocument, extractUploadIdentities, + matchUploadIdentities, type UploadIdentityExtractor, type UploadPreflightQuery, } from '@rawback/sdk' @@ -32,6 +32,7 @@ import { } from './sftp-client.ts' import { formatBytes } from './ui/format.ts' import { findBundledExiftoolPath } from './upload-identity.ts' +import { ledgerPathFor, UploadAcknowledgements } from './upload-ledger.ts' import { DEFAULT_UPLOAD_STATE_PATH, UploadState, type UploadStateFile } from './upload-state.ts' export const SUPPORTED_UPLOAD_EXTENSIONS = new Set([ @@ -63,10 +64,14 @@ export interface UploadCommandOptions { concurrency: number dryRun: boolean path: string + /** Ignore what this computer remembers uploading and ask the server about every file. */ + recheck?: boolean } export interface UploadCommandDependencies extends ReadCommandDependencies { client?: RawbackClient + /** Defaults to `upload-ledger.jsonl` beside the upload state. */ + ledgerPath?: string sleep?: (milliseconds: number) => Promise statePath?: string transportFactory?: UploadTransportFactory @@ -85,12 +90,15 @@ export interface UploadFile { interface UploadPreflight { account: string + /** Part of the ledger scope, with the numeric user id. */ + apiHost: string client: RawbackClient endpoint: string files: UploadFile[] password: string quotaBytes: number usedBytes: number + userId: number | string username: string } @@ -99,7 +107,6 @@ interface UploadFailure { file: UploadFile } -const REMOTE_QUERY_BATCH_SIZE = 200 const FALLBACK_BYTES_PER_SECOND = 10_000_000 / 8 function canonicalEndpoint(value: string): string { @@ -261,48 +268,43 @@ async function preflight( return { account: account.slug, + apiHost: client.http.apiHost, client, endpoint: canonicalEndpoint(sftp.endpoint), files: await scanUploadPath(options.path), password: sftp.password, quotaBytes: Number(account.storageQuotaBytes), usedBytes: Number(account.storageUsedBytes), + userId: account.id, username: sftp.username, } } +/** The server's image id for each file it already holds, keyed by client key. */ async function remoteIdentities( client: RawbackClient, files: UploadFile[], dependencies: UploadCommandDependencies, -): Promise> { - const result = new Set() +): Promise> { const candidates = files.filter( (file): file is UploadFile & { capturedAt: string; clientKey: string } => Boolean(file.capturedAt && file.clientKey), ) try { - for (let index = 0; index < candidates.length; index += REMOTE_QUERY_BATCH_SIZE) { - const identities = candidates.slice(index, index + REMOTE_QUERY_BATCH_SIZE).map((file) => ({ + return await matchUploadIdentities( + client.graphql, + candidates.map((file) => ({ clientKey: file.clientKey, originalFilename: file.basename, capturedAt: file.capturedAt, - })) - const response = await client.graphql.query({ - query: ExistingUploadIdentitiesDocument, - variables: { identities }, - }) - if (response.error) throw response.error - if (!response.data) throw new Error('Remote duplicate check did not return image data') - for (const match of response.data.existingUploadIdentities) result.add(match.clientKey) - } + })), + ) } catch { commandOutput(dependencies).warning( 'Remote duplicate checking is unavailable; continuing with SFTP verification.', ) - result.clear() + return new Map() } - return result } async function identifyFiles( @@ -310,6 +312,7 @@ async function identifyFiles( dependencies: UploadCommandDependencies, metadataConcurrency?: number, ): Promise { + if (files.length === 0) return [] const sidecarPath = await findBundledExiftoolPath() const extractor = dependencies.identityExtractor ?? extractUploadIdentities const keyed = files.map((file) => ({ file, clientKey: crypto.randomUUID() })) @@ -398,6 +401,7 @@ async function uploadRound( dependencies: UploadCommandDependencies, progress: UploadProgressController, shouldStop: () => boolean, + uploaded: UploadFile[], ): Promise<{ failures: UploadFailure[]; uploadedBytes: number }> { const failures: UploadFailure[] = [] let uploadedBytes = 0 @@ -414,6 +418,7 @@ async function uploadRound( progress.update(file, bytes) }) state.setFileStatus(persisted, 'completed') + uploaded.push(file) uploadedBytes += file.size progress.complete(file) } catch (error) { @@ -459,21 +464,50 @@ export async function runUpload( const state = options.dryRun ? await UploadState.openReadonly(statePath) : await UploadState.open(statePath) + const acks = await UploadAcknowledgements.open({ + path: dependencies.ledgerPath ?? ledgerPathFor(statePath), + apiHost: preflightResult.apiHost, + accountId: preflightResult.userId, + warn: (message) => ui.warning(message), + }) try { - const identified = collapseLocalIdentities( - await identifyFiles(preflightResult.files, dependencies, environment.metadata?.concurrency), - ) - const locallyCompleted = new Set( - state - ? identified - .filter((file) => state.isCompleted(stateFile(preflightResult, file))) - .map((file) => file.canonicalPath) - : [], - ) - const localPending = identified.filter((file) => !locallyCompleted.has(file.canonicalPath)) + // What this computer already uploaded is settled before any metadata is + // read, so re-selecting a mostly uploaded card costs a stat per file. + const review = options.recheck + ? { skipped: [], recheck: [], unknown: preflightResult.files } + : acks.review(preflightResult.files) + let alreadyUploaded = review.skipped.length + const identified: UploadFile[] = [...review.recheck] + for (const file of await identifyFiles( + review.unknown, + dependencies, + environment.metadata?.concurrency, + )) { + if (!options.recheck && acks.skipByIdentity(file)) { + alreadyUploaded += 1 + } else if ( + // The upload-state resume record predates the ledger. Like a transfer + // acknowledgement, it is only trusted for photos without a capture + // time; the rest go to the server's check, and are recorded from there. + !options.recheck && + !file.capturedAt && + state?.isCompleted(stateFile(preflightResult, file)) + ) { + acks.record(file, 'transfer', 'upload') + alreadyUploaded += 1 + } else { + identified.push(file) + } + } + const localPending = collapseLocalIdentities(identified) const remote = await remoteIdentities(preflightResult.client, localPending, dependencies) - const pending = localPending.filter((file) => !file.clientKey || !remote.has(file.clientKey)) + const pending: UploadFile[] = [] + for (const file of localPending) { + const imageId = file.clientKey ? remote.get(file.clientKey) : undefined + if (imageId === undefined) pending.push(file) + else acks.record(file, 'server', 'remote', imageId) + } const pendingBytes = pending.reduce((sum, file) => sum + file.size, 0) const remainingQuota = Math.max(0, preflightResult.quotaBytes - preflightResult.usedBytes) if (pendingBytes > remainingQuota) { @@ -492,7 +526,7 @@ export async function runUpload( const source = estimate?.source ?? '10 Mbps fallback' ui.document( uploadDryRunDocument({ - completed: locallyCompleted.size, + alreadyUploaded, estimatedSeconds: pendingBytes / rate, existingRemote: remote.size, files: pending.length, @@ -505,8 +539,9 @@ export async function runUpload( } if (!state) throw new Error('Unable to open upload progress database') + await acks.flush() if (pending.length === 0) { - ui.info('Nothing to upload; all supported files are already complete or remote.') + ui.info('Nothing to upload; all supported files are already uploaded or remote.') return } @@ -548,6 +583,7 @@ export async function runUpload( let totalUploaded = 0 const progress = new UploadProgressController(pendingBytes, pending.length, ui) const permanentFailures: UploadFailure[] = [] + const uploaded: UploadFile[] = [] for (let round = 0; round < 3 && remaining.length > 0 && !interrupted; round += 1) { const result = await uploadRound( remaining, @@ -558,6 +594,7 @@ export async function runUpload( dependencies, progress, () => interrupted, + uploaded, ) totalUploaded += result.uploadedBytes const reconnectable = result.failures.filter(({ error }) => isConnectionFailure(error)) @@ -580,6 +617,12 @@ export async function runUpload( ) } await progress.finish() + const confirmation = await acks.acknowledge(preflightResult.client.graphql, uploaded) + if (confirmation.unconfirmed > 0) { + ui.warning( + `Rawback did not confirm ${confirmation.unconfirmed} uploaded photo${confirmation.unconfirmed === 1 ? '' : 's'}; ${confirmation.unconfirmed === 1 ? 'it' : 'they'} will be uploaded again next time.`, + ) + } for (const failure of permanentFailures) { state.setFileStatus( stateFile(preflightResult, failure.file), @@ -597,8 +640,10 @@ export async function runUpload( ui.document( uploadSummaryDocument({ cancelled: interrupted, + confirmed: confirmation.confirmed, failedFiles: permanentFailures.length, totalBytes: totalUploaded, + unconfirmed: confirmation.unconfirmed, }), ) if (interrupted) throw new Error('Upload cancelled') diff --git a/test/cli.test.ts b/test/cli.test.ts index b335e68..e24f3b2 100644 --- a/test/cli.test.ts +++ b/test/cli.test.ts @@ -236,6 +236,8 @@ describe('rawback photos upload CLI', () => { expect(result.stdout).toContain('--concurrency') expect(result.stdout).toContain('parallel SFTP transfers') expect(result.stdout).toContain('--dry-run') + expect(result.stdout).toContain('--recheck') + expect(result.stdout).toContain('ask the server about every') }) test('requires an upload path before running preflight', () => { diff --git a/test/upload.test.ts b/test/upload.test.ts index 7cdc32e..6ab28c4 100644 --- a/test/upload.test.ts +++ b/test/upload.test.ts @@ -1,9 +1,9 @@ import { afterEach, describe, expect, test } from 'bun:test' -import { chmod, mkdtemp, rm, symlink } from 'node:fs/promises' +import { chmod, mkdtemp, realpath, rm, symlink } from 'node:fs/promises' import { tmpdir } from 'node:os' import { basename, join } from 'node:path' -import type { UploadIdentityExtractor } from '@rawback/sdk' +import { UploadLedger, uploadLedgerScope, type UploadIdentityExtractor } from '@rawback/sdk' import type { RawbackClient } from '../src/client.ts' import type { @@ -85,11 +85,24 @@ function operationName(query: unknown): string { return document.definitions[0]?.name?.value ?? '' } +const API_HOST = 'https://api.rawback.app' +const SCOPE = uploadLedgerScope({ apiHost: API_HOST, accountId: 1 }) + +/** + * A server holding the photos named in `remote`. Pass a `Set` to share it with + * a {@link RecordingTransport}, which adds what it uploads — the gateway + * recording a photo before it answers the close. + */ function fakeClient( - remote: string[] = [], - options: { credentials?: boolean; enabled?: boolean; duplicateQueryFails?: boolean } = {}, + remote: string[] | Set = [], + options: { + credentials?: boolean + enabled?: boolean + duplicateQueryFails?: boolean + identityQueries?: string[][] + } = {}, ): RawbackClient { - const remoteSet = new Set(remote) + const remoteSet = remote instanceof Set ? remote : new Set(remote) return { config: {}, credentials: @@ -118,6 +131,7 @@ function fakeClient( } if (options.duplicateQueryFails) return { error: new Error('query unavailable') } const identities = request.variables?.identities ?? [] + options.identityQueries?.push(identities.map(({ originalFilename }) => originalFilename)) return { data: { existingUploadIdentities: identities @@ -127,7 +141,7 @@ function fakeClient( } }, }, - http: {}, + http: { apiHost: API_HOST }, } as unknown as RawbackClient } @@ -146,7 +160,10 @@ class RecordingTransport implements UploadTransport { maxActive = 0 uploads: string[] = [] - constructor(private readonly fail = new Set()) {} + constructor( + private readonly fail = new Set(), + private readonly library?: Set, + ) {} async connect(): Promise { this.connected += 1 @@ -165,6 +182,7 @@ class RecordingTransport implements UploadTransport { onProgress(information.size) await Bun.sleep(15) if (this.fail.has(basename(remotePath))) throw new Error('rejected by test server') + this.library?.add(basename(remotePath)) } finally { this.active -= 1 } @@ -301,10 +319,13 @@ describe('upload command', () => { }, ) - expect(lines[0]).toContain('Files 1') - expect(lines[0]).toContain('Remote 1') + expect(lines[0]).toContain('Files 1') + expect(lines[0]).toContain('Remote 1') expect(lines[0]).toContain('10 Mbps fallback') await expect(Bun.file(statePath).stat()).rejects.toMatchObject({ code: 'ENOENT' }) + await expect(Bun.file(join(directory, 'upload-ledger.jsonl')).stat()).rejects.toMatchObject({ + code: 'ENOENT', + }) }) test('uses configured metadata concurrency during upload review', async () => { @@ -417,3 +438,261 @@ describe('upload command', () => { expect(errors.some((message) => message.includes('Failed bad.jpg'))).toBe(true) }) }) + +async function readLedger(path: string): Promise { + const ledger = new UploadLedger({ path }) + await ledger.load() + return ledger +} + +function countingExtractor(seen: string[]): UploadIdentityExtractor { + return async (candidates, options) => { + seen.push(...candidates.map(({ originalFilename }) => originalFilename)) + return identityExtractor(candidates, options) + } +} + +describe('upload acknowledgements', () => { + test('confirms uploads and skips them next time without reading metadata or asking', async () => { + const directory = await temporaryDirectory() + const configPath = await writeConfig(directory) + await Bun.write(join(directory, 'one.jpg'), 'one') + await Bun.write(join(directory, 'two.jpg'), 'two') + const library = new Set() + const transport = new RecordingTransport(new Set(), library) + const lines: string[] = [] + const base = { + configPath, + statePath: join(directory, 'upload-state.json'), + stdout: (message: string) => lines.push(message), + transportFactory: factoryFor(transport), + } + + await runUpload( + { concurrency: 2, dryRun: false, path: directory }, + { ...base, client: fakeClient(library), identityExtractor }, + ) + + expect(transport.uploads.sort()).toEqual(['one.jpg', 'two.jpg']) + expect(lines.at(-1)).toContain('Confirmed 2') + const ledger = await readLedger(join(directory, 'upload-ledger.jsonl')) + expect(ledger.size).toBe(2) + expect(ledger.findByIdentity(SCOPE, { name: 'one.jpg', capturedAt })).toMatchObject({ + ack: 'server', + source: 'upload', + imageId: expect.any(Number), + }) + + const extracted: string[] = [] + const identityQueries: string[][] = [] + await runUpload( + { concurrency: 2, dryRun: false, path: directory }, + { + ...base, + client: fakeClient(library, { identityQueries }), + identityExtractor: countingExtractor(extracted), + }, + ) + + expect(transport.uploads).toHaveLength(2) + expect(extracted).toEqual([]) + expect(identityQueries).toEqual([]) + expect(lines.at(-1)).toContain('Nothing to upload') + }) + + test('uploads again a photo the server did not record', async () => { + const directory = await temporaryDirectory() + const configPath = await writeConfig(directory) + await Bun.write(join(directory, 'lost.jpg'), 'lost') + const transport = new RecordingTransport() + const warnings: string[] = [] + const dependencies = { + client: fakeClient(), + configPath, + identityExtractor, + statePath: join(directory, 'upload-state.json'), + stderr: (message: string) => warnings.push(message), + stdout: () => undefined, + transportFactory: factoryFor(transport), + } + + await runUpload({ concurrency: 1, dryRun: false, path: directory }, dependencies) + expect(warnings.some((message) => message.includes('did not confirm 1 uploaded photo'))).toBe( + true, + ) + expect((await readLedger(join(directory, 'upload-ledger.jsonl'))).size).toBe(0) + + await runUpload({ concurrency: 1, dryRun: false, path: directory }, dependencies) + expect(transport.uploads).toEqual(['lost.jpg', 'lost.jpg']) + }) + + test('asks the server again about a photo whose confirmation failed', async () => { + const directory = await temporaryDirectory() + const configPath = await writeConfig(directory) + await Bun.write(join(directory, 'photo.jpg'), 'photo') + const library = new Set() + const transport = new RecordingTransport(new Set(), library) + const warnings: string[] = [] + const base = { + configPath, + statePath: join(directory, 'upload-state.json'), + stderr: (message: string) => warnings.push(message), + stdout: () => undefined, + transportFactory: factoryFor(transport), + } + + await runUpload( + { concurrency: 1, dryRun: false, path: directory }, + { ...base, client: fakeClient(library, { duplicateQueryFails: true }), identityExtractor }, + ) + expect(warnings.some((message) => message.includes('could not confirm'))).toBe(true) + expect( + (await readLedger(join(directory, 'upload-ledger.jsonl'))).findByIdentity(SCOPE, { + name: 'photo.jpg', + capturedAt, + })?.ack, + ).toBe('transfer') + + const extracted: string[] = [] + const identityQueries: string[][] = [] + await runUpload( + { concurrency: 1, dryRun: false, path: directory }, + { + ...base, + client: fakeClient(library, { identityQueries }), + identityExtractor: countingExtractor(extracted), + }, + ) + + // The capture time comes from the ledger; the server is asked, and has it. + expect(extracted).toEqual([]) + expect(identityQueries).toEqual([['photo.jpg']]) + expect(transport.uploads).toEqual(['photo.jpg']) + expect( + (await readLedger(join(directory, 'upload-ledger.jsonl'))).findByIdentity(SCOPE, { + name: 'photo.jpg', + capturedAt, + }), + ).toMatchObject({ ack: 'server', source: 'remote' }) + }) + + test('recognises a moved photo by identity and records its new path', async () => { + const directory = await temporaryDirectory() + const configPath = await writeConfig(directory) + const card = join(directory, 'card') + const archive = join(directory, 'archive') + await Bun.write(join(card, 'photo.jpg'), 'photo') + await Bun.write(join(archive, 'photo.jpg'), 'photo') + const library = new Set() + const transport = new RecordingTransport(new Set(), library) + const identityQueries: string[][] = [] + const dependencies = { + client: fakeClient(library, { identityQueries }), + configPath, + identityExtractor, + statePath: join(directory, 'upload-state.json'), + stdout: () => undefined, + transportFactory: factoryFor(transport), + } + + await runUpload({ concurrency: 1, dryRun: false, path: card }, dependencies) + identityQueries.splice(0) + await runUpload({ concurrency: 1, dryRun: false, path: archive }, dependencies) + + expect(transport.uploads).toEqual(['photo.jpg']) + expect(identityQueries).toEqual([]) + const ledger = await readLedger(join(directory, 'upload-ledger.jsonl')) + expect( + ledger.findByFile(SCOPE, { + canonicalPath: await realpath(join(archive, 'photo.jpg')), + size: 5, + mtimeMs: (await Bun.file(join(archive, 'photo.jpg')).stat()).mtimeMs, + }), + ).toMatchObject({ ack: 'server' }) + }) + + test('--recheck ignores the ledger and asks the server about every file', async () => { + const directory = await temporaryDirectory() + const configPath = await writeConfig(directory) + await Bun.write(join(directory, 'photo.jpg'), 'photo') + const library = new Set() + const transport = new RecordingTransport(new Set(), library) + const extracted: string[] = [] + const identityQueries: string[][] = [] + const dependencies = { + client: fakeClient(library, { identityQueries }), + configPath, + identityExtractor: countingExtractor(extracted), + statePath: join(directory, 'upload-state.json'), + stdout: () => undefined, + transportFactory: factoryFor(transport), + } + + await runUpload({ concurrency: 1, dryRun: false, path: directory }, dependencies) + extracted.splice(0) + identityQueries.splice(0) + await runUpload({ concurrency: 1, dryRun: false, path: directory, recheck: true }, dependencies) + + expect(extracted).toEqual(['photo.jpg']) + expect(identityQueries).toEqual([['photo.jpg']]) + expect(transport.uploads).toEqual(['photo.jpg']) + }) + + test('remembers a photo without a capture time once its upload succeeds', async () => { + const directory = await temporaryDirectory() + const configPath = await writeConfig(directory) + await Bun.write(join(directory, 'export.png'), 'png') + const transport = new RecordingTransport() + const noCaptureTime: UploadIdentityExtractor = async (candidates) => ({ + identities: [], + uncheckedClientKeys: candidates.map(({ clientKey }) => clientKey), + failedClientKeys: [], + }) + const dependencies = { + client: fakeClient(), + configPath, + identityExtractor: noCaptureTime, + // A fresh state file each run: the ledger alone has to remember it. + stdout: () => undefined, + transportFactory: factoryFor(transport), + ledgerPath: join(directory, 'ledger.jsonl'), + } + + await runUpload( + { concurrency: 1, dryRun: false, path: directory }, + { ...dependencies, statePath: join(directory, 'first.json') }, + ) + await runUpload( + { concurrency: 1, dryRun: false, path: directory }, + { ...dependencies, statePath: join(directory, 'second.json') }, + ) + + expect(transport.uploads).toEqual(['export.png']) + expect((await readLedger(join(directory, 'ledger.jsonl'))).size).toBe(1) + }) + + test('keeps uploading when the ledger cannot be read', async () => { + const directory = await temporaryDirectory() + const configPath = await writeConfig(directory) + await Bun.write(join(directory, 'photo.jpg'), 'photo') + const transport = new RecordingTransport() + const warnings: string[] = [] + + await runUpload( + { concurrency: 1, dryRun: false, path: join(directory, 'photo.jpg') }, + { + client: fakeClient(), + configPath, + identityExtractor, + ledgerPath: directory, + statePath: join(directory, 'upload-state.json'), + stderr: (message: string) => warnings.push(message), + stdout: () => undefined, + transportFactory: factoryFor(transport), + }, + ) + + expect(transport.uploads).toEqual(['photo.jpg']) + expect(warnings.some((message) => message.includes('could not be read'))).toBe(true) + }) +})