@yeez-tech/dianshu-file-transfer

Dianshu data trading platform — secure file download engine

Overview

@yeez-tech/dianshu-file-transfer provides a resumable pipeline for downloading and decrypting files from the Dianshu platform:

Download → Validate → PrefetchKey → SubmitChainTx → FetchKey → Decrypt

Built on a generic pipeline engine (ResumableWorkflow) with pause / resume / persist / error-recovery for each step.

Features

• Streaming decrypt — sealed .crdownload → plaintext output, SHA256-verifiable
• Pause & resume — at any stage: download, chain tx polling, decryption
• Chain transaction — automated off-chain-skey submission and signing
• i18n — progress-text and error messages in English & Chinese
• Event-driven UI — progress, progress-text, action-error, complete, stop
• Task lifecycle — removeTask({ deleteFile }) cleans all state: workflow, ydl, DB, disk

Design boundary: what the package does not do

Compared with the Dianshu website client, the package starts after downloadList / checksumUrl are already available. Obtaining them is the caller's responsibility — this is intentional, not a missing feature.

Caller must finish before newDownloadTask

| Step | What | Notes | |------|------|-------| | 1 | Create task (putTask, etc.) | Out of package | | 2 | Send transaction A requestOffChain | Triggers backend to produce the result file | | 3 | Poll order until taskReady=1 | Read downloadList / checksumUrl from order API |

Only then call the package with those URLs. The package then:

download file → requestOffChainSkey (tx B) → wait for decrypt key → decrypt

Transaction B (requestOffChainSkey, for the decryption key) is handled inside the package. You do not need to send every on-chain tx before calling the package — only transaction A (produce file) + a ready download list.

Required options at call time

| Field | Meaning | |-------|---------| | downloadList / checksumUrl | Ready download / verify URLs (from order after tx A) | | privateKey / publicKey | User YPC private key + sealing public key | | filePath / downloadFileName | Output path + temp-file basename | | taskOrderId | 数字 task id（taskList 返回的 id）→ getTaskPrivateKey | | taskOrderCode | 字符串 taskCode（taskList 返回的 taskCode）→ checkTask、链上交易 B、downloadNotify | | taskKey (newDownloadTask 第 1 参) | 任意 workflow 键，与 taskOrderId / taskOrderCode 无关 |

Why this split: the package is a download + decrypt engine; creating orders, signing tx A, and UI/private-key prompts belong to the host app. Details: docs/download-divergence-01-downloadList.md §§1–6.

Storage namespace (dataPath)

One dataPath (one dsft.db) ↔ one live FileDownloader at a time.

Isolation is by UserContext.dataPath, not by a constructor namespace option. Multiple concurrent tasks on a single FileDownloader are fine; two FileDownloaders pointing at the same dataPath is unsupported (misuse). See docs §7.1.

Installation

npm install @yeez-tech/dianshu-file-transfer

Requirements: Node.js ≥ 18. Browser is not supported (uses fs, path, worker threads).

Quick start

import { initDSFT, FileDownloader } from "@yeez-tech/dianshu-file-transfer";
import { AppContext } from "@yeez-tech/dianshu-file-transfer/context/AppContext";
import { UserContext } from "@yeez-tech/dianshu-file-transfer/context/UserContext";

// 1. One-time library init (ydl worker threads, i18n)
await initDSFT({ language: "zh-CN" });

// 2. Contexts
const appCtx  = new AppContext("https://api.example.com");
const userCtx = new UserContext({
  token:     "your-auth-token",   // ← keep updated on refresh!
  dataPath:  "/path/to/data",
  cachePath: "/path/to/cache",
});

// 3. Downloader（可选覆盖 FetchKey：checkTask 轮询 / 重试 / 超时）
const dl = new FileDownloader(appCtx, userCtx, {
  fetchKeyPoll: {
    intervalMs: 1_000,            // default（对齐客户端）
    maxPrepareRetries: 3,         // publishStatus=3 重发交易上限
    maxChainWaitMs: 600_000,      // 等上链终态超时 → CHAIN_TX_TIMEOUT
    maxPrivateKeyWaitMs: 600_000, // status=2 后等私钥超时 → PRIVATE_KEY_FETCH_TIMEOUT
  },
});

// 4. Start — downloadList / checksumUrl must already be ready
//    (caller finished: create task → tx A requestOffChain → poll order)
await dl.newDownloadTask("task-001", {
  filePath:         "/downloads/my-dataset.csv",
  privateKey:       "ypc-private-key-hex",
  publicKey:        "sealing-public-key-hex",
  downloadFileName: "my-dataset",
  taskOrderId:          1234,
  taskOrderCode:        "P1709696493",
  downloadList: {
    "0": [{ url: "https://cdn.example.com/file.sealed" }],
  },
  checksumUrl:      "https://cdn.example.com/file.checksum",
}, "your-auth-token");

// 5. Lifecycle
await dl.pauseTask("task-001");
await dl.resumeTask("task-001", "your-auth-token");  // awaits completion

// After crash / force-quit (same dataPath):
// const dl2 = new FileDownloader(appCtx, userCtx);
// dl2.markInterruptedPaused(); // optional; aligns with website login
// await dl2.resumeTask("task-001", freshToken);

await dl.removeTask("task-001", { deleteFile: true });
await dl.close(); // pauseAll + tear down ydl

API reference

initDSFT(options?)

One-time init. Call once before creating any FileDownloader.

await initDSFT({ language: "en" | "zh-CN" });  // default "en"

FileDownloader

new FileDownloader(appCtx, userCtx, options?)

| Option | Notes | |--------|-------| | fetchKeyPoll.intervalMs | Poll interval (default 1000) | | fetchKeyPoll.maxPrepareRetries | Resend tx B when publishStatus=3 (default 3) | | fetchKeyPoll.maxChainWaitMs | Wait for on-chain success before timeout (default 600_000) → CHAIN_TX_TIMEOUT | | fetchKeyPoll.maxPrivateKeyWaitMs | After publishStatus=2, wait for key (default 600_000) → PRIVATE_KEY_FETCH_TIMEOUT |

PrefetchKey (poll:false) is unaffected. FetchKey polls checkTask / publishStatus (client ChainPollingAction), then getTaskPrivateKey after status 2.

newDownloadTask(taskKey, options, token?)

Start a download workflow. taskKey must be unique.

| Option | Required | Notes | |--------|----------|-------| | filePath | ✅ | Final output path | | privateKey | ✅ | YPC private key (hex) | | publicKey | ✅ | Sealing public key (hex) | | downloadFileName | ✅ | Basename for temp files | | taskOrderId | ✅ | Numeric id from POST /system/task/taskList | | taskOrderCode | ✅ | taskCode from same row (tx B / checkTask / notify) | | downloadList | ✅ | { priority: { url }[] } | | checksumUrl | ✅ | ydl VerificationFile URL | | datasetVersion | — | 1 = skip app hash; 2 = require encryptFileHash | | encryptFileHash | if v2 | Sealed-file hash from order API |

taskKey (1st arg to newDownloadTask) is an arbitrary workflow id — not the numeric order id.

pauseTask(taskKey)

Pause the active workflow. Emits progress-text with stage-specific text. If there is no live workflow (e.g. after a crash), still marks the task paused so it can be resumed later.

resumeTask(taskKey, token)

Resume a paused or crash-interrupted incomplete task. Awaits full completion (success or error). Accepts tasks whose paused flag is still false after a force-kill (website can also call markInterruptedPaused() on login first — both paths work). Always pass a fresh token (CDN URL refresh).

listIncompleteTasks() / markInterruptedPaused()

• listIncompleteTasks() — !isSuccess && !isError (for UI / startup).
• markInterruptedPaused() — set paused: true on those rows without starting them (aligns with website resetTaskStatus / force-quit accounting).

Persisted DownloadTaskData.process is overall pipeline progress in [0, 1] (updated during download/decrypt). After interrupt, read downloadTaskDB.get(taskKey)?.process for the order-page progress bar — no host-side task JSON required.

removeTask(taskKey, options?)

Remove all task state:

• Stop & cleanup workflow (calls each Action.cleanup(context))
• Remove from ydl Downloader (memory + progressIO)
• Delete pipeline persistent state (Sqlite)
• Delete from DownloadTaskDataDB
• If { deleteFile: true }: deletes output + cache files

await dl.removeTask("task-001", { deleteFile: true });

pauseAll() / close()

pauseAll stops all active workflows and marks incomplete tasks paused. close calls pauseAll then tears down the ydl Downloader — use on graceful app quit. After a crash, reconstruct FileDownloader with the same dataPath, optionally markInterruptedPaused(), then resumeTask(key, token).

Token management

UserContext.token is used for all authenticated API calls (DSAPI injects it as a header). If the token is refreshed elsewhere, update userContext.token to keep downloads working after the old token expires.

userCtx.token = newAccessToken;

Events

All events are emitted with { taskKey, ... } so the UI can filter per task.

| Event | Payload | Fires when | |-------|---------|------------| | progress | { taskKey, current, total } | Pipeline progress update | | progress-text | { taskKey, text } | Status bar text — i18n-aware (see table below) | | task-download-speed | { taskKey, speed } | Bytes/s from ydl | | action-error | { taskKey, actionName, error } | A specific Action failed — actionName identifies the stage | | error | { taskKey, error } | Any pipeline error | | complete | { taskKey, result } | Workflow finished successfully | | stop | { taskKey, info } | Workflow was stopped |

Action names in action-error

| actionName | Stage | |-------------|-------| | Download | DownloadAction | | Validate | ValidateAction (app-level sealed hash when datasetVersion===2) | | PrefetchKey | FetchPrivateKeyAction (poll:false) | | SubmitChainTx | SubmitChainTxAction | | FetchKey | FetchPrivateKeyAction (poll:true) | | Decrypt | DecryptAction |

progress-text locale reference

| Key | en | zh-CN | |-----|-----|-------| | DOWNLOAD_PAUSED | Download paused | 下载已暂停 | | DOWNLOAD_RESUMED | Download resumed | 下载已恢复 | | DECRYPT_PAUSED | Decryption paused | 解密已暂停 | | DECRYPT_RESUMED | Decryption resumed | 解密已恢复 | | FETCH_KEY_PAUSED | Key fetching paused | 密钥获取已暂停 | | FETCH_KEY_RESUMED | Key fetching resumed | 密钥获取已恢复 | | SUBMIT_TX_PAUSED | Waiting for transaction… | 交易提交无法取消… | | FETCH_KEY_STARTED | Fetching off-chain private key | 正在获取链下私钥 | | FETCH_KEY_WAITING | Waiting for off-chain private key… | 等待链下私钥下发… | | FETCH_KEY_SUCCEEDED | Off-chain private key obtained | 获取链下私钥成功 | | FETCH_KEY_FAILED | Failed to obtain off-chain private key: {reason} | 获取链下私钥失败：{reason} | | FETCH_KEY_PREFETCH_MISS | Off-chain key not ready; will submit chain tx then keep waiting | 暂无链下私钥，将提交链上交易后继续等待 | | PRIVATE_KEY_FETCH_TIMEOUT | Timed out waiting for decryption private key | 解密私钥获取超时 | | CHAIN_TX_TIMEOUT | On-chain transaction timed out | 上链交易超时 | | CHAIN_TX_MAX_RETRIES | On-chain transaction retry limit reached | 上链交易重试次数已达上限 | | SUBMIT_TX_STARTED | Submitting chain transaction | 正在提交链上交易 | | SUBMIT_TX_SUCCEEDED | Chain transaction submitted | 链上交易已提交 | | SUBMIT_TX_SKIPPED | Chain transaction already submitted, skipping | 链上交易已提交，跳过 | | SUBMIT_TX_SKIP_CHAIN_STATUS | On-chain status present, skipping submit | 链上已有交易状态，跳过发送 | | DECRYPT_STARTED | Decryption started | 开始解密 | | DECRYPT_SUCCEEDED | Decryption succeeded | 解密成功 |

i18n

import { setLocale } from "@yeez-tech/dianshu-file-transfer/utils/i18n";
setLocale("zh-CN");   // Chinese
setLocale("en");      // English (default)

Architecture

FileDownloader
 └─ ResumableWorkflow (persistable via SqliteWorkflowStorageAdapter)
     ├─ DownloadAction           (ydl Downloader)
     ├─ ValidateAction           (app-level sealed hash if datasetVersion===2)
     ├─ FetchPrivateKeyAction    (PrefetchKey, poll:false)
     ├─ SubmitChainTxAction      (init → dryRun → sign → send, tx B)
     ├─ FetchPrivateKeyAction    (FetchKey, poll:true — checkTask/publishStatus → key)
     └─ DecryptAction            (RecoverableReadStream → Unsealer → RecoverableWriteStream)

Each PipelineAction has a unified lifecycle: execute / stop / resume / cleanup.

Logging

This library uses loglevel.

import log from "@yeez-tech/dianshu-file-transfer/utils/logger";
log.setLevel("warn");  // silence info/debug

See the loglevel docs for remote logging plugins (Sentry, etc.).

License

MIT