@agstack/storage-file

Enterprise-grade async file storage plugin for AGStack

A reference implementation for every AGStack storage plugin. Receives completed transactions from the Runtime through the Plugin SDK and persists them to the filesystem using a fully asynchronous, non-blocking pipeline.

Features

• Async Pipeline — Queue → Worker → Buffer → Batch Writer → Rotation → Compression → Retention
• Multiple Formats — JSON, JSONL, NDJSON, CSV, Compressed JSON, Compressed NDJSON
• File Rotation — Daily, Hourly, Weekly, Monthly, Size-based, Record-count
• Compression — gzip and brotli in background workers
• Batch Writing — Configurable batch size, interval, memory threshold
• Buffer Management — Memory buffer with disk spill, backpressure, overflow protection
• Retention Policy — 7/30/90/180/365 days, custom, auto-cleanup, archive before delete
• Crash Recovery — Partial write detection, marker files, disk buffer recovery
• Concurrency — Multiple workers, file locking, race condition prevention
• Security — Path traversal protection, symlink detection, sensitive field masking
• Health Check — Disk usage, write speed, pending batches, error tracking
• Configurable Naming — Hostname, environment, app name, PID, custom templates

Installation

npm install @agstack/storage-file

Quick Start

import { StorageFilePlugin, DEFAULTS, validateConfig } from "@agstack/storage-file";
import { createRuntime } from "@agstack/logger";

const runtime = createRuntime({
  plugins: [
    () => {
      const plugin = new StorageFilePlugin();
      runtime.registerPlugin("storage-file", plugin, {
        options: validateConfig({
          storagePath: "logs",
          format: "jsonl",
          batchSize: 100,
          flushIntervalMs: 2000,
          rotation: "daily",
          retention: "30d",
        }),
      });
      return plugin;
    },
  ],
});

await runtime.start();

Configuration

| Option | Type | Default | Description | |--------|------|---------|-------------| | storagePath | string | "logs" | Root storage directory | | format | StorageFormat | "jsonl" | File format (json, jsonl, ndjson, csv, compressed-json, compressed-ndjson) | | compression | "gzip" \| "brotli" \| "none" | "none" | Compression algorithm | | compressionLevel | number | 6 | Compression level (1-9 for gzip, 1-11 for brotli) | | batchSize | number | 100 | Records per batch | | maxBatchSize | number | 1000 | Maximum batch size | | flushIntervalMs | number | 5000 | Flush interval in milliseconds | | maxBufferSize | number | 10000 | Maximum records in memory buffer | | memoryThresholdBytes | number | 67108864 | Memory threshold for emergency flush (64MB) | | rotation | RotationPolicy | "daily" | Rotation policy | | maxFileSizeBytes | number | 104857600 | Maximum file size before rotation (100MB) | | maxRecordsPerFile | number | 100000 | Maximum records per file | | retention | RetentionPeriod | "30d" | Retention period | | retentionDays | number | 30 | Custom retention days | | archiveBeforeDelete | boolean | true | Archive files before deletion | | workerCount | number | 2 | Number of background workers | | maskFields | string[] | (sensitive fields) | Fields to mask in payloads |

API

StorageFilePlugin

| Method | Description | |--------|-------------| | save(transaction) | Queue a transaction for async writing | | saveBatch(transactions) | Queue multiple transactions | | flush() | Force-flush all pending writes | | health() | Get plugin health status |

health() response

{
  status: "healthy" | "degraded" | "unavailable";
  metrics: {
    uptimeMs: number;
    transactionsProcessed: number;
    queueSize: number;
    errorRate: number;
  };
  errors?: Array<{ code: string; message: string; recovered: boolean }>;
  warnings?: Array<{ code: string; message: string }>;
}

Architecture

Runtime → Queue → Worker → Buffer → Batch Writer → Rotation → Compression → File
                         ↕
                     Disk Buffer
                         ↕
                    Crash Recovery

Directory Structure

logs/
├── YYYY/
│   ├── MM/
│   │   ├── DD/
│   │   │   ├── YYYY-MM-DD.jsonl
│   │   │   └── ...
│   │   └── ...
│   └── ...
├── archive/
│   └── ...
└── .buffer/
    └── ...

Error Handling

The plugin never blocks request threads. Errors are captured and reported through:

• Event bus (storage.saved, storage.failed)
• Health check metrics
• Logger integration
• Automatic retry with exponential backoff

Crash Recovery

On startup, the plugin:

1. Scans for marker files from interrupted writes
2. Validates partial files and recovers valid records
3. Cleans up disk buffer spill files
4. Reports recovery status

Security

• Path traversal prevention
• Symlink attack detection
• Sensitive field masking (passwords, tokens, API keys, etc.)
• Sanitized filenames
• Configurable allowed paths

License

MIT