@codaco/ 网络出口商

运行时间不可知 效应-TS 向GraphML 和 CSV 输出 Canvas 的访谈会。 主机插入了三层( H)InterviewRepository, (中文(简体) ), ProtocolRepository, (中文(简体) ), Output)和打电话 exportPipeline。该软件包没有持久性,没有 node:* 在节点、浏览器和Cloudflare Workers上运行.

安装

effect 是一个对等依赖,所以在软件包旁边安装它。 效果4还没有实现稳定的释放,所以通过标签要求释放候选人:

npm install @codaco/network-exporters effect@rc

import { Effect, Layer, Queue } from 'effect';
import { exportPipeline } from '@codaco/network-exporters/pipeline';
import type { ExportEvent } from '@codaco/network-exporters/events';
import { makeZipOutput } from '@codaco/network-exporters/layers/ZipOutput';

const layer = Layer.mergeAll(
  MyInterviewRepository, // workspace-specific
  MyProtocolRepository, // workspace-specific
  makeZipOutput(mySink), // ships with this package
);

const program = Effect.gen(function* () {
  const queue = yield* Queue.unbounded<ExportEvent>();
  return yield* exportPipeline(interviewIds, exportOptions, queue);
});

const result = await Effect.runPromise(program.pipe(Effect.provide(layer)));
// → { status, successfulExports, failedExports, output }

为什么是包裹

导出逻辑是历史上嵌入的 弗雷斯科 和其他网络Canvas应用程序。 此软件包将管道提取为自成一体,主机不可知库,这样:

• 多个应用程序可以共享相同的导出语义和文件格式.
• 管道可隔离测试(没有Prisma,没有S3,没有临时文件).
• 数据库、协议存储和输出目的地可通过Effect Layers连接.
• 流到端 AsyncIterable<Uint8Array> 整个节点的最低共同分母 Readable网络 ReadableStream和工人——所以同一个包裹在每一个运行时间运行.

建筑

┌──────────────────── consumer (host application) ─────────────────────┐
│                                                                       │
│  Provides Layers for these Tags ─────────────┐                        │
│                                              ▼                        │
│              ┌───────────────────── exportPipeline ─────────────────┐ │
│              │                                                     │ │
│              │  fetching → formatting → generating → outputting    │ │
│              │     ▲           ▲             ▲            ▲        │ │
│              │     │           │             │            │        │ │
│   Interview ─┘  Protocol     pure         per-format    Output     │ │
│   Repository    Repository  (in-process)   fan-out      lifecycle  │ │
│                                                       (begin/write/│ │
│                                                          end)      │ │
│              └─────────────────────────────────────────────────────┘ │
│                                              │                        │
│                                              ▼                        │
│                            ExportReturn { status,                     │
│                                           successfulExports,          │
│                                           failedExports,              │
│                                           output: OutputResult }      │
└───────────────────────────────────────────────────────────────────────┘

3个注入服务(Context.Services) 构成包的输入/输出边界:

通过散列排列的会话参考协议,所以管道在每个运行时都会完全获取每个独特的协议一次,不管有多少会话共享. 这个 Output 服务状态 : begin 生产不透明的手柄,管道每成功生成一个条目 end 返回自定义的主机 OutputResult 流出通过 ExportReturn.output。 。 。 .

东道主的唯一责任是:

1. 执行三个服务层.
2. 打电话 exportPipeline(ids, options, queue) 和表面进展事件.

Sparse 属性和稳定的输出计划

格式化阶段分析每个会话网络 NcNetworkSchema 在此之前 任何 CSV 或 GraphML 物质都收到。 没有新输入项 属性,而定义的值如 false, (中文(简体) ), 0, (中文(简体) ), '',以及 [] 留下来. 无效的定义值产生 session-processing 失败 会话; 它不中止运行中的其他会话 .

输出计划既来自协议编码本,也来自稀有的实体数据:

• 代码表声明保持完全未解变量的可见. 这样一来 文件结构稳定,而不是取决于对参与者的提问 恰好接了电话.
• 没有可适用的代码簿声明的被定义的实体属性是 外出属性. 它们留在CSV和GraphML中而不是 丢弃.
• 缺失仅控制数据输出; 它不会去除已声明的列或 键.

简历

节点和边缘 CSV 文件被代表的实体类型分割. 每个文件 标题包括为该类型声明的每个变量,即使变量 每行都缺席。 Ego CSV 信头同样包括每一个宣称的自负 变量。 当一个会话有零节点或零边缘时,仅以页眉为主的文件 使用代码簿中的所有节点或边缘定义.

分类变量扩展为每个已声明选项一列,并布局 变量扩展以协调列。 一个缺失值会产生空单元格 . 定义的空绝对数组产生显式 false 选项单元格,所以 仍可区别于未解的绝对变量。 现 外部属性使用其属性ID来添加由数据驱动的列.

图表ML

GraphML 在代码簿中声明所有自负、节点和边缘变量的密钥, 包括属于无人任职实体类型的变量和与 零节点或零边缘。 因此,一个完全未答复的已宣布变量保留 联合国 <key> 但表示没有 <data> 键.

当前的外部属性得到无法确定的散列密钥ID 与代码簿变量ID相撞. 整个密钥分配一次 文档 <data> 元素使用分配的ID。 如果同一个外部 属性出现在多个实体类型上, GraphML 释放出一个 for="all" 键,而不是相冲突的节点和边缘键.

公众表面

进口使用子路径——该套件没有桶出口.

其他一切(formaters,会话帮助器,调度逻辑,内部zip流帮助器)都是内部的,而不是导出.

使用量

1. 提供 InterviewRepository

import { Effect, Layer } from 'effect';
import { NcNetworkSchema } from '@codaco/shared-consts';
import { DatabaseError } from '@codaco/network-exporters/errors';
import type { InterviewExportInput } from '@codaco/network-exporters/input';
import { InterviewRepository } from '@codaco/network-exporters/services/InterviewRepository';

export const PrismaInterviewRepository = Layer.succeed(InterviewRepository, {
  getForExport: (ids) =>
    Effect.gen(function* () {
      const rows = yield* Effect.tryPromise({
        try: () =>
          prisma.interview.findMany({
            where: { id: { in: [...ids] } },
            include: { participant: true },
          }),
        catch: (error) => new DatabaseError({ cause: error }),
      });

      const inputs: InterviewExportInput[] = rows.map((row) => ({
        id: row.id,
        participantIdentifier: row.participant.identifier,
        startTime: row.startTime,
        finishTime: row.finishTime,
        network: NcNetworkSchema.parse(row.network),
        protocolHash: row.protocolHash,
      }));

      return inputs;
    }),
});

存储器返回引用协议的会话 通过散列没有 protocol: true 加入——管道通过 ProtocolRepository。主机验证网络 NcNetworkSchema.parse() 直接; 货包不再装载 a parseNcNetwork 帮手.

2. 提供a ProtocolRepository

import { Effect, Layer } from 'effect';
import { DatabaseError } from '@codaco/network-exporters/errors';
import type { ProtocolExportInput } from '@codaco/network-exporters/input';
import { ProtocolRepository } from '@codaco/network-exporters/services/ProtocolRepository';

export const PrismaProtocolRepository = Layer.succeed(ProtocolRepository, {
  getProtocols: (hashes) =>
    Effect.gen(function* () {
      const rows = yield* Effect.tryPromise({
        try: () =>
          prisma.protocol.findMany({ where: { hash: { in: [...hashes] } } }),
        catch: (error) => new DatabaseError({ cause: error }),
      });

      const map: Record<string, ProtocolExportInput> = {};
      for (const row of rows) {
        map[row.hash] = {
          hash: row.hash,
          name: row.name,
          codebook: row.codebook,
        };
      }
      return map;
    }),
});

输油管呼叫 getProtocols 使用从会话列表中提取的已调试的散列集运行一次。 Hashes 主持人不返回 被处理为缺失 —— 引用它们的会话路由 failedExports 与 kind: "protocol-missing" 而不是中止运行.

3. 提供 Output

Output 替换旧的 FileStorage + 键 FileSystem 具有单一状态生命周期的标记 。 货船 makeZipOutput(sink) 对于想要今天捆绑-zip行为的主机: 它通过纯 JS 流出条目 fflate 并转发所产生的zip字节(作为 AsyncIterable<Uint8Array>)给一个主机提供的汇回调. 不希望拉链执行的主机 Output 直接说.

S3为 ZipOutput (节点)

import { Readable } from 'node:stream';
import { Upload } from '@aws-sdk/lib-storage';
import { Effect } from 'effect';
import { OutputError } from '@codaco/network-exporters/errors';
import { makeZipOutput } from '@codaco/network-exporters/layers/ZipOutput';

const S3ZipOutput = makeZipOutput((stream, fileName) =>
  Effect.tryPromise({
    try: async () => {
      await new Upload({
        client: s3,
        params: { Bucket, Key: fileName, Body: Readable.from(stream) },
      }).done();
      return { key: fileName, url: await presign(fileName) };
    },
    catch: (cause) => new OutputError({ cause }),
  }),
);

Readable.from(asyncIterable) 调整软件包 AsyncIterable<Uint8Array> 转到节点 Readable aWS SDK 的预期. @aws-sdk/lib-storage是因为 Upload 处理多段上传和每段内部重试.

浏览器 blob 与 ZipOutput

import { Effect } from 'effect';
import { OutputError } from '@codaco/network-exporters/errors';
import { makeZipOutput } from '@codaco/network-exporters/layers/ZipOutput';

const BlobZipOutput = makeZipOutput((stream) =>
  Effect.tryPromise({
    try: async () => {
      const chunks: Uint8Array[] = [];
      for await (const chunk of stream) chunks.push(chunk);
      const blob = new Blob(chunks, { type: 'application/zip' });
      return { blob, url: URL.createObjectURL(blob) };
    },
    catch: (cause) => new OutputError({ cause }),
  }),
);

fflate 纯JS,所以这个层工作 在浏览器和Cloudflare工人 没有多填充.

OPFS 文件夹( 无拉链)

import { Effect, Layer } from 'effect';
import { OutputError } from '@codaco/network-exporters/errors';
import { Output } from '@codaco/network-exporters/services/Output';

const OPFSFolderOutput = Layer.succeed(Output, {
  begin: () =>
    Effect.tryPromise({
      try: async () => ({
        dir: await navigator.storage
          .getDirectory()
          .then((root) =>
            root.getDirectoryHandle(`export-${Date.now()}`, { create: true }),
          ),
      }),
      catch: (cause) => new OutputError({ cause }),
    }),
  writeEntry: (handle, entry) =>
    Effect.tryPromise({
      try: async () => {
        const file = await handle.dir.getFileHandle(entry.name, {
          create: true,
        });
        const writable = await file.createWritable();
        for await (const chunk of entry.data) await writable.write(chunk);
        await writable.close();
      },
      catch: (cause) => new OutputError({ cause }),
    }),
  end: (handle) => Effect.succeed({ folderHandle: handle.dir }),
});

此执行跳过 makeZipOutput 完全: 每一个成功的生成结果都写为一个 OPFS 文件, 返回 OutputResult 将目录控件带回主机。 它拼写成相同的 Output 标记 —— 管道中没有特殊标记 .

4. 运行管道

import { type Cause, Effect, Fiber, Layer, Queue, Stream } from 'effect';
import { exportPipeline } from '@codaco/network-exporters/pipeline';
import type { ExportEvent } from '@codaco/network-exporters/events';
import {
  describeExportError,
  type ExportError,
} from '@codaco/network-exporters/errors';

const exportLayer = Layer.mergeAll(
  PrismaInterviewRepository,
  PrismaProtocolRepository,
  S3ZipOutput,
);

const result = await Effect.scoped(
  Effect.gen(function* () {
    // `Cause.Done` in the error type is what lets `Queue.end` close the queue
    // below: `Queue.end` takes an `Enqueue<A, E | Done>`.
    const queue = yield* Queue.unbounded<ExportEvent, Cause.Done>();

    // Drain the queue into the UI (e.g. an SSE response). Hold the fiber:
    // finishing the export is not the same as having delivered every event,
    // and the join below is what makes delivery a property rather than a
    // matter of how the scheduler happened to interleave.
    const drain = yield* Effect.forkScoped(
      Stream.fromQueue(queue).pipe(Stream.runForEach(renderEventToClient)),
    );

    const pipelineResult = yield* exportPipeline(
      ['interview-1', 'interview-2'],
      {
        exportGraphML: true,
        exportCSV: true,
        globalOptions: {
          useScreenLayoutCoordinates: true,
          screenLayoutHeight: 1080,
          screenLayoutWidth: 1920,
        },
        // optional: defaults to os.cpus().length on Node, 4 elsewhere
        concurrency: 4,
        // optional: only used to populate session metadata
        appVersion: '3.0.0',
        commitHash: process.env.COMMIT_HASH,
      },
      queue,
    );

    // `end` lets the buffered events drain and then terminates the stream;
    // `Queue.shutdown` would discard whatever is still buffered.
    yield* Queue.end(queue);
    yield* Fiber.join(drain);

    return pipelineResult;
  }),
).pipe(
  Effect.provide(exportLayer),
  Effect.catch((error: ExportError) =>
    Effect.succeed({
      status: 'error' as const,
      message: describeExportError(error, 'running export'),
    }),
  ),
  Effect.runPromise,
);

result.output 随心所欲 Output.end 返回执行 S3ZipOutput 上面,这是 { key, url }没有 zipUrl 或者说 zipKey 字段打开 ExportReturn; 想要读取 URL 的主机 result.output.url。 。 。 .

管道阶段

四个阶段,四个阶段 stage 事件值. outputting 覆盖 begin,每个 writeEntry,以及 end. 管道没有持续状态——没有单独的清理阶段.

货币

generateOutputFilesEffect 粉丝跨出交叉产品(会场×格式×分型). 汇兑方式为: ExportOptions.concurrency默认为 os.cpus().length 在节点上(通过防守的动态导入解决)和在其他地方的小型固定值。 为内存受约束的运行时间(小服务器功能)设定更低,为基准设定更高.

{ ...options, concurrency: 2 }

这个 outputting 阶段本身是相接的——一个拉链流,按进入顺序写出.

错误模式

两个明显的错误轨道, 都标记了 .

致命错误 — 效果失败通道

管道升入无法恢复条件的 Effect 失败通道: 数据库获取失败, 输出开始/ 写入/ 结束失败 。 每个都有标记 Data.TaggedError 实例:

这个 ExportError 工会类型涵盖两个致命的班级.

内部失败 Output.writeEntry 是致命的——一旦部分字节在主机的捆绑中,重试内线是主机的工作,而不是管道的工作. 需要重试语义的主机在其内部执行 Output 层; fflate 失败和主机回调失败 OutputError,则原始错误可用于 error.cause。 。 。 .

describeExportError(error, stage?) 产生人能读取的信息。 它发送标签和检查 error.cause 用于已知运行时模式( q)code === "ENOSPC", OOM 消息, 超时, 连接被拒绝), 然后再返回到标签意识默认状态 。 在消费边界上使用——包本身从不构建用户化的字符串.

Effect.catch((error) =>
  Effect.succeed({
    status: 'error' as const,
    message: describeExportError(error, 'running export'),
  }),
);

部分失败—— ExportFailure[]

单个错误会话或单个失败文件确实存在 没有 中止输油管。 三样东西可以降落在 failedExports数字 :

成功的文件仍然被写入 Output 并报告.

const result = await Effect.runPromise(...);
//   ↓ ExportReturn
// {
//   status: "success" | "partial",
//   successfulExports: ExportSuccess[],
//   failedExports:     ExportFailure[],
//   output:            OutputResult,   // host-defined
// }

if (result.status === "partial") {
    for (const failure of result.failedExports) {
        switch (failure.kind) {
            case "protocol-missing":
            case "session-processing":
            case "generation":
                console.warn(describeExportError(failure.error));
                break;
        }
    }
}

status 实值 "partial" 任何时候 failedExports.length > 0不管它属于哪种人 这种分解——致命-Vs-部分——是故意的:消费者可以给用户一个可用的捆绑,即使一个CSV失败,同时仍然可以表面显示丢失的东西.

进展活动

exportPipeline 写入 ExportEvents 到您提供的队列。 两个事件形状:

type ExportStageEvent = {
  type: 'stage';
  stage: 'fetching' | 'formatting' | 'generating' | 'outputting';
  message: string;
};

type ExportProgressEvent = {
  type: 'progress';
  stage: 'generating' | 'outputting';
  current: number;
  total: number;
};

每一次入境 progress 事件在两个阶段中都释放出来 generating (每个文件)和 outputting (每个文件编写). 绘制栏的当前 UI 代码 generating 工作情况不变 outputting - 形状是一样的,只有 stage 值不同.

软件包只发出这两种事件类型——其他任何类型(例如SSE形状) complete编号 :error 客户端交付的事件)是主机的关注.

stageMessages 导出为 a Record<ExportStage, string> 预先本地化的英语字符串; 消费者可能忽略它并释放自己的字符串.

地方发展

这个包裹住在 network-canvas-monorepo 并遵循标准脚本.

# from monorepo root
pnpm --filter @codaco/network-exporters build       # vite build
pnpm --filter @codaco/network-exporters test        # vitest --run
pnpm --filter @codaco/network-exporters test:watch
pnpm --filter @codaco/network-exporters typecheck   # tsc --noEmit

内部消费者决心 package.json#exports 直接反对 src/*.ts ——不需要建设步骤来取出源变化. 这个 build 脚本仍然运行 vite build,该输出 dist/<entry>.{js,d.ts} 定义的每个公共子路径 vite.config.ts;该产出仅由已公布npm文物消耗,通过 package.json#publishConfig.exports 在打包时间交换。 全部运行时间依赖关系( E)effect, (中文(简体) ), fflate, (中文(简体) ), @codaco/shared-consts, (中文(简体) ), @codaco/protocol-validation, (中文(简体) ), es-toolkit, (中文(简体) ), sanitize-filename, (中文(简体) ), zod, (中文(简体) ), @xmldom/xmldom, (中文(简体) ), ohash)在构建时被外部化. effect 是一个同行而不是依赖,所以在同样直接使用Effect的应用程序中,完全有一个副本——和一个纤维运行时间.

添加新的切入点

1. 添加源文件于 src/<path>.ts。 。 。 .
2. 将条目添加到 vite.config.ts build.lib.entry (出版物的需要) dist/ 建楼).
3. 将相应的子路径添加到 package.json#exports,指着新的 src/<path>.ts 文档.
4. 将匹配的子路径添加到 package.json#publishConfig.exports,指着 ./dist/<path>.js编号 :.d.ts因此,出版的文物 不断揭露它.
5. 运行 pnpm --filter @codaco/network-exporters build 并验证匹配 dist/<path>.js + 键 .d.ts 土地(这只验证了公布的文物——内部消费者已经通过下列途径看到新的条目: src/) (中文(简体) ).

添加运行时间依赖

喜欢工作空间目录。 如果 Dep 已经有一个目录条目 pnpm-workspace.yaml,将其称为 "<dep>": "catalog:"。否则,在目录中添加一个被绑定的版本,然后引用它。 一定要把它外部化进去 vite.config.ts 所以不是捆绑的 浏览器/工人兼容性表示避免 node:* 从任何新代码导入 src/ —— 说 node:os 在被监视的动态导入中 session/generateOutputFiles.ts 这是唯一的例外.

版本

此软件包使用 更改就像其他的单体一样 在修改用户后 :

pnpm changeset

选择 @codaco/network-exporters 并选择凸起类型。 别撞了 package.json#version 手边.

许可证

GPL-3.0-或-后传. 见 LICENSE 在单发根.