APFS — Autoprocessing File System

apfs-io/apfs is an automated file-processing system built for high-performance, pipeline-driven handling of complex objects. Upload a file and a declarative workflow takes over: validating, transforming, and persisting every derived artifact while tracking progress in real time.

Features

• Declarative workflows — GitHub Actions-inspired YAML defines your processing pipeline per bucket.
• Pre-upload validation — synchronous size and content-type checks before any file is stored.
• DAG execution — jobs declare needs: dependencies; independent jobs run in parallel on matched workers.
• Conditional logic — if: expressions skip jobs based on upstream outputs or statuses.
• Failure policies — per-job on-failure: fail | continue | retry:N controls pipeline behaviour.
• Processing state — every object carries a ProcessingState (progress 0–1, per-job/step statuses).
• Complex objects — a single upload may produce many derived files (thumb.jpg, 720p.mp4, …).
• gRPC + REST gateway — Protocol Buffers API with an auto-generated REST facade.
• Pluggable storage drivers — local filesystem and S3-compatible backends out of the box.
• Pluggable event streams — NATS, Kafka, or Redis pub/sub for object events and processing status.

Quick Start

import (
    "github.com/apfs-io/apfs/libs/client"
    "github.com/apfs-io/apfs/models"
)

cl, _ := client.Connect("localhost:8080")
videos := cl.Group("videos")

// 1. Define the processing workflow for the bucket (once per deployment).
wf := &models.Workflow{
    Version:      "2",
    ContentTypes: []string{"video/*"},
    Validate: &models.WorkflowValidate{
        MaxSize:      "2GB",
        ContentTypes: []string{"video/mp4", "video/quicktime"},
    },
    Jobs: map[string]*models.WorkflowJob{
        "thumbnail": {
            RunsOn: "cpu",
            Steps: []*models.WorkflowStep{
                {Uses: "ffmpeg/thumbnail", With: map[string]any{"target": "thumb.jpg", "time": "00:00:02"}},
            },
        },
        "transcode-720p": {
            RunsOn: "gpu",
            Needs:  []string{"thumbnail"},
            Steps: []*models.WorkflowStep{
                {Uses: "ffmpeg/encode", With: map[string]any{"target": "720p.mp4", "resolution": "1280x720"}},
            },
        },
    },
}
videos.SetWorkflow(ctx, wf)

// 2. Upload a file — validation runs synchronously, processing starts asynchronously.
f, _ := os.Open("promo.mp4")
obj, _ := videos.Upload(ctx, f, client.WithTags("promo"))

// 3. Poll progress.
videos.WatchProgress(ctx, obj.ID, func(state *models.ProcessingState) {
    fmt.Printf("progress: %.0f%%  status: %s\n", state.Progress*100, state.Status)
})

See docs/USE_CASES.md for complete examples and docs/WORKFLOW.md for the full schema reference.

Deployment & initialization

On startup APFS can automatically configure bucket workflows from a directory (default /workflows):

/workflows/{groupName}/manifest.yaml

Mount the directory in Docker or bake manifests into the image — no separate seed container is needed. See docs/INITIALIZATION.md for the full startup sequence, environment variables, and upgrade rules.

# Local stack (workflows mounted from deploy/workflows)
make build-docker-dev && make run

# Ubuntu server: Docker + Redis + systemd under /opt/apfs
curl -fsSL https://raw.githubusercontent.com/apfs-io/apfs/main/deploy/standalone/install.sh | sudo bash

See deploy/README.md for standalone options (APFS_IMAGE, APFS_PREFIX).

Event streams

Object lifecycle events and optional processing-status updates are published over a pluggable pub/sub backend. Set EVENTSTREAM_CONNECT and, if needed, PROCESSING_STATUS_STREAM_CONNECT to one of:

Scheme	Example
nats://	nats://nats:4222/apfs?topics=events
kafka://	kafka://broker1:9092,broker2:9092/group?topics=events
redis://	redis://localhost:6379/0?topics=events
rediss://	rediss://user:pass@localhost:6379/0?topics=events (TLS)

Channels come from the topics query parameter. See docs/PROCESSING.md for status-stream subscription from a client.

API Overview

Services

1. Object operations

   • Head — retrieve object metadata and ProcessingState.
   • Get — fetch an object's data stream and metadata.
   • Refresh — trigger re-processing of an existing object.
   • Delete — remove an object or specific sub-files.

2. Workflow management

   • SetWorkflow — store or update the processing workflow for a bucket.
   • GetWorkflow — retrieve the current workflow for a bucket.

3. Data upload

   • Upload — stream a new file into the system; pre-upload validation runs before persistence.

Processing State

Every object carries a ProcessingState describing the execution of its workflow:

{
  "object_id": "videos/abc123",
  "status": "running",
  "progress": 0.33,
  "jobs": {
    "thumbnail": { "status": "completed", "outputs": { "path": "thumb.jpg" } },
    "transcode-720p": { "status": "running", "worker": "gpu-node-1" },
    "transcode-480p": { "status": "pending" }
  }
}

Possible top-level statuses: pending, running, completed, partial (some jobs failed with on-failure: continue), failed.

REST Endpoints

Method	Endpoint	Description
GET	/v1/head/{id}	Retrieve object metadata.
GET	/v1/object/{id}	Retrieve object and data stream.
PUT	/v1/refresh/{id}	Trigger re-processing of an object.
PUT	/v1/workflow/{group}	Set the workflow for a bucket.
GET	/v1/workflow/{group}	Retrieve the workflow for a bucket.
POST	/v1/object	Upload a new file.
DELETE	/v1/object/{id}	Delete an object or specific sub-files.

Protocol Buffers

The API is defined with proto3 in protocol/v1/. A REST gateway is generated via gRPC-Gateway. OpenAPI specs are available for easy client generation.

Documentation

Document	Description
docs/INITIALIZATION.md	Service startup, workflow bootstrap, Docker deployment
docs/PROCESSING.md	Processing status lifecycle and event-stream subscription
docs/WORKFLOW.md	Full v2 workflow YAML schema reference
docs/USE_CASES.md	End-to-end examples: image gallery, video, documents, avatars
deploy/README.md	Docker images, compose stack, example manifests
internal/driver/s3/README.md	Local S3 setup with MinIO

License

This project is licensed under the MIT License. See the LICENSE file for details.