|
|
|
@@ -0,0 +1,57 @@
|
|
|
|
|
import { randomUUID } from 'node:crypto';
|
|
|
|
|
import { Inject, Injectable } from '@nestjs/common';
|
|
|
|
|
import type { CreateCollectionItemRequest, ListCollectionItemsQuery, UpdateCollectionItemRequest } from '@atlas/contracts';
|
|
|
|
|
import type { Database, DatabaseTransaction } from '@atlas/database';
|
|
|
|
|
import { ApiError } from '../common/api-error.js';
|
|
|
|
|
import { IdempotencyService } from '../common/idempotency.service.js';
|
|
|
|
|
import { DATABASE } from '../database/database.module.js';
|
|
|
|
|
|
|
|
|
|
type Row = { id: string; product_id: string; source_kind: 'platform' | 'manual'; platform_code: string | null; source_url: string | null; source_title: string | null; source_min_price: string | null; source_max_price: string | null; source_currency_code: string | null; note: string | null; status: string; created_at: Date | string; updated_at: Date | string; revision: number; variant_count: string; transfer_status: string | null };
|
|
|
|
|
@Injectable()
|
|
|
|
|
export class CollectionService {
|
|
|
|
|
public constructor(@Inject(IdempotencyService) private readonly idempotency: IdempotencyService, @Inject(DATABASE) private readonly database: Database) {}
|
|
|
|
|
async list(workspaceId: string, platformId: string, query: ListCollectionItemsQuery) {
|
|
|
|
|
const cursor = query.cursor ? decodeCursor(query.cursor) : null;
|
|
|
|
|
return this.database.workspaceTransaction(workspaceId, async (tx) => {
|
|
|
|
|
await this.requireHeaderPlatform(tx, platformId);
|
|
|
|
|
const rows = await tx.query<Row>(`${selectRow}
|
|
|
|
|
WHERE c.deleted_at IS NULL AND ($1::text IS NULL OR c.source_title ILIKE '%' || $1 || '%')
|
|
|
|
|
AND ($2::text IS NULL OR p.code = $2) AND ($3::text IS NULL OR $3 = 'not_listed')
|
|
|
|
|
AND ($4::timestamptz IS NULL OR (c.updated_at, c.id) < ($4, $5::uuid))
|
|
|
|
|
${collectionGroupBy} ORDER BY c.updated_at DESC, c.id DESC LIMIT $6`, [query.q ?? null, query.source_platform_code ?? null, query.listing_status ?? null, cursor?.at ?? null, cursor?.id ?? null, query.limit + 1]);
|
|
|
|
|
const items = rows.rows.slice(0, query.limit).map(toItem); const last = items.at(-1);
|
|
|
|
|
const count = await tx.query<{ count: string; collection_item_limit: number }>('SELECT count(*)::text AS count, max(w.collection_item_limit)::int AS collection_item_limit FROM atlas.collection_items c CROSS JOIN atlas.workspaces w WHERE c.workspace_id = w.id AND c.deleted_at IS NULL');
|
|
|
|
|
return { data: items, meta: { has_more: rows.rows.length > query.limit, next_cursor: rows.rows.length > query.limit && last ? encodeCursor(last.updated_at, last.id) : null, active_count: Number(count.rows[0]?.count ?? 0), collection_limit: count.rows[0]?.collection_item_limit ?? 1000 } };
|
|
|
|
|
});
|
|
|
|
|
}
|
|
|
|
|
async create(workspaceId: string, userId: string, key: string, input: CreateCollectionItemRequest) {
|
|
|
|
|
return this.idempotency.run({ workspaceId, userId, method: 'POST', path: '/api/collection-items', key, request: input }, async (tx) => {
|
|
|
|
|
const item = await tx.query<{ id: string }>(`INSERT INTO atlas.collection_items (workspace_id, source_kind, source_item_key, source_title, note, status) VALUES (current_setting('app.workspace_id')::uuid, 'manual', $1, $2, $3, 'ready_for_edit') RETURNING id`, [randomUUID(), input.title ?? null, input.note ?? null]);
|
|
|
|
|
const product = await tx.query<{ id: string }>(`INSERT INTO atlas.products (workspace_id, collection_item_id, source_title) VALUES (current_setting('app.workspace_id')::uuid, $1, $2) RETURNING id`, [item.rows[0].id, input.title ?? null]);
|
|
|
|
|
await tx.query(`INSERT INTO atlas.product_variants (workspace_id, product_id, display_name, sort_order) VALUES (current_setting('app.workspace_id')::uuid, $1, '默认 SKU', 0)`, [product.rows[0].id]);
|
|
|
|
|
return { status: 201, body: { data: { collection_item: await this.requireItem(tx, item.rows[0].id), product_id: product.rows[0].id, editor_entry_path: `/collection-items/${item.rows[0].id}/edit` } } };
|
|
|
|
|
});
|
|
|
|
|
}
|
|
|
|
|
async update(workspaceId: string, userId: string, id: string, key: string, expected: number, input: UpdateCollectionItemRequest) {
|
|
|
|
|
return this.idempotency.run({ workspaceId, userId, method: 'PATCH', path: `/api/collection-items/${id}`, key, request: { if_match: expected, ...input } }, async (tx) => {
|
|
|
|
|
const result = await tx.query(`UPDATE atlas.collection_items SET note = $1, revision = revision + 1, updated_at = now() WHERE id = $2 AND deleted_at IS NULL AND revision = $3`, [input.note, id, expected]);
|
|
|
|
|
if (!result.rowCount) await this.versionConflict(tx, id, expected);
|
|
|
|
|
return { status: 200, body: { data: await this.requireItem(tx, id) } };
|
|
|
|
|
});
|
|
|
|
|
}
|
|
|
|
|
async bulkDelete(workspaceId: string, userId: string, key: string, ids: { ids: string[] }) { return this.removeMany(workspaceId, userId, key, '/api/collection-items/bulk-delete', ids, false); }
|
|
|
|
|
async clear(workspaceId: string, userId: string, key: string, input: { confirm: true; expected_active_count: number }) { return this.removeMany(workspaceId, userId, key, '/api/collection-items/clear', input, true); }
|
|
|
|
|
async retryAssets(workspaceId: string, userId: string, id: string, key: string, input: { transfer_ids: string[] }) { return this.idempotency.run({ workspaceId, userId, method: 'POST', path: `/api/collection-items/${id}/asset-retries`, key, request: input }, async (tx) => { await this.requireItem(tx, id); const result = await tx.query(`UPDATE atlas.collection_asset_transfers SET status = 'queued', error_payload = NULL, updated_at = now() WHERE collection_item_id = $1 AND id = ANY($2::uuid[]) AND status IN ('failed', 'cancelled')`, [id, input.transfer_ids]); if (result.rowCount !== input.transfer_ids.length) throw new ApiError(422, 'ASSET_RETRY_INVALID', '只能重试当前采集记录中失败或已取消的素材任务'); const operation = await this.operation(tx, userId, 'collection_asset_retry', id); await this.event(tx, id, 'collection.assets.retry.requested', { operation_id: operation.id, collection_item_id: id }); return { status: 202, body: { data: { operation } } }; }); }
|
|
|
|
|
private async removeMany(workspaceId: string, userId: string, key: string, path: string, input: unknown, clear: boolean) { return this.idempotency.run({ workspaceId, userId, method: 'POST', path, key, request: input }, async (tx) => { const requested = input as { ids?: string[]; expected_active_count?: number }; const count = await tx.query<{ count: string }>('SELECT count(*)::text AS count FROM atlas.collection_items WHERE deleted_at IS NULL'); if (clear && Number(count.rows[0].count) !== requested.expected_active_count) throw new ApiError(409, 'COLLECTION_COUNT_CHANGED', '采集箱数量已变化，请重新确认'); const deleted = await tx.query<{ id: string }>(clear ? `UPDATE atlas.collection_items SET status = 'deleted', deleted_at = now(), revision = revision + 1, updated_at = now() WHERE deleted_at IS NULL RETURNING id` : `UPDATE atlas.collection_items SET status = 'deleted', deleted_at = now(), revision = revision + 1, updated_at = now() WHERE id = ANY($1::uuid[]) AND deleted_at IS NULL RETURNING id`, clear ? [] : [requested.ids]); if (!clear && deleted.rows.length !== requested.ids?.length) throw new ApiError(404, 'RESOURCE_NOT_FOUND', '部分采集记录不存在'); const operation = await this.operation(tx, userId, 'collection_cleanup', deleted.rows[0]?.id ?? null); await this.event(tx, operation.id, 'collection.cleanup.requested', { operation_id: operation.id, collection_item_ids: deleted.rows.map((row) => row.id) }); return { status: 202, body: { data: { deleted_count: deleted.rows.length, operation } } }; }); }
|
|
|
|
|
private async operation(tx: DatabaseTransaction, userId: string, kind: string, resourceId: string | null) { const row = await tx.query<{ id: string; status: string; created_at: Date | string }>(`INSERT INTO atlas.operations (workspace_id, kind, resource_type, resource_id, created_by_user_id) VALUES (current_setting('app.workspace_id')::uuid, $1, 'collection_item', $2, $3) RETURNING id, status, created_at`, [kind, resourceId, userId]); return { id: row.rows[0].id, kind, status: row.rows[0].status, created_at: iso(row.rows[0].created_at), completed_at: null, error: null }; }
|
|
|
|
|
private async event(tx: DatabaseTransaction, aggregateId: string, type: string, payload: unknown) { await tx.query(`INSERT INTO atlas.outbox_events (workspace_id, aggregate_type, aggregate_id, event_type, event_payload, deduplication_key) VALUES (current_setting('app.workspace_id')::uuid, 'collection_item', $1, $2, $3, $4)`, [aggregateId, type, payload, `${type}:${aggregateId}`]); }
|
|
|
|
|
private async requireHeaderPlatform(tx: DatabaseTransaction, id: string) { const row = await tx.query(`SELECT id FROM atlas.platforms WHERE id = $1 AND status = 'active' AND coalesce((capabilities->>'can_be_header_context')::boolean, false)`, [id]); if (!row.rows[0]) throw new ApiError(422, 'PLATFORM_UNAVAILABLE', '当前平台不可用于 ERP 上下文'); }
|
|
|
|
|
private async requireItem(tx: DatabaseTransaction, id: string) { const row = await tx.query<Row>(`${selectRow} WHERE c.id = $1 AND c.deleted_at IS NULL ${collectionGroupBy}`, [id]); if (!row.rows[0]) throw new ApiError(404, 'RESOURCE_NOT_FOUND', '采集记录不存在'); return toItem(row.rows[0]); }
|
|
|
|
|
private async versionConflict(tx: DatabaseTransaction, id: string, expected: number) { const row = await tx.query<{ revision: number }>('SELECT revision FROM atlas.collection_items WHERE id = $1 AND deleted_at IS NULL', [id]); if (!row.rows[0]) throw new ApiError(404, 'RESOURCE_NOT_FOUND', '采集记录不存在'); if (row.rows[0].revision !== expected) throw new ApiError(409, 'VERSION_CONFLICT', '采集记录已被更新', [{ field: 'If-Match', code: 'VERSION_CONFLICT', message: `当前版本为 ${row.rows[0].revision}`, current_revision: row.rows[0].revision }]); }
|
|
|
|
|
}
|
|
|
|
|
const selectRow = `SELECT c.id, product.id AS product_id, c.source_kind, p.code AS platform_code, c.source_url, c.source_title, c.source_min_price::text, c.source_max_price::text, c.source_currency_code, c.note, c.status, c.created_at, c.updated_at, c.revision, count(DISTINCT v.id)::text AS variant_count, min(t.status) AS transfer_status FROM atlas.collection_items c JOIN atlas.products product ON product.collection_item_id = c.id LEFT JOIN atlas.platforms p ON p.id = c.source_platform_id LEFT JOIN atlas.product_variants v ON v.product_id = product.id AND v.deleted_at IS NULL LEFT JOIN atlas.collection_asset_transfers t ON t.collection_item_id = c.id`;
|
|
|
|
|
const collectionGroupBy = 'GROUP BY c.id, product.id, p.code';
|
|
|
|
|
function toItem(row: Row) { return { id: row.id, product_id: row.product_id, source: { kind: row.source_kind, platform_code: row.platform_code, url: row.source_url }, title: row.source_title, thumbnail: { asset_id: null, preview_url: null, status: row.transfer_status ?? 'none' }, source_price: { min: row.source_min_price, max: row.source_max_price, currency_code: row.source_currency_code }, variant_summary: { count: Number(row.variant_count), needs_review: row.status === 'needs_review' }, note: row.note, collection_status: row.status, listing_summary: { status: 'not_listed' as const, detail: '等待首次上架', last_error: null }, created_at: iso(row.created_at), updated_at: iso(row.updated_at), revision: row.revision }; }
|
|
|
|
|
function iso(value: Date | string) { return value instanceof Date ? value.toISOString() : value; }
|
|
|
|
|
function encodeCursor(at: string, id: string) { return Buffer.from(JSON.stringify({ at, id })).toString('base64url'); }
|
|
|
|
|
function decodeCursor(cursor: string): { at: string; id: string } { try { const value = JSON.parse(Buffer.from(cursor, 'base64url').toString('utf8')) as { at?: string; id?: string }; if (!value.at || !value.id || Number.isNaN(Date.parse(value.at))) throw new Error(); return { at: value.at, id: value.id }; } catch { throw new ApiError(422, 'VALIDATION_FAILED', '分页游标无效'); } }
|
R524809