morpheus

What It Does

Morpheus transforms data. It gives you:

• 50+ built-in transformers — string, numeric, datetime, validation, normalization, privacy, schema, aggregation
• YAML-driven rule engine — define transformation logic in config, not code
• DAG pipeline orchestration — multi-step transformations with parallel execution
• PII detection and masking — email, SSN, phone, credit card pattern detection with masking
• Format conversion — JSON, YAML, XML bidirectional conversion
• Stream parsing — token-based JSON, XML, CSV parsing

It uses hexagonal architecture (ports and adapters), so everything is behind interfaces and testable.

What It Is Not

This is not an application framework. It does not replace your HTTP server, database layer, message broker, or deployment infrastructure. It is a library you import and call.

Features

Data Transformers

50+ transformers across 10 categories. Each implements the Transformer interface. Same API, composable, chainable.

Rule Engine

Define transformation rules in YAML. The engine compiles them, indexes fields, and evaluates data against conditions.

DAG Pipeline

Multi-step transformation pipelines where stages can depend on each other. Independent stages run in parallel automatically.

PII Detection and Masking

Regex-based detection for email, SSN, phone, credit card, passport. Configurable masking strategies (partial, full, hash). AES-256 tokenization with reversible detokenization. Audit logging for compliance.

Caching

In-memory (LRU with TTL), Redis, or multi-level (L1 memory + L2 Redis). Stampede protection. Cache warming. Redis adapters support standalone, cluster, sentinel, and ring topologies via redis.UniversalClient.

Resilience

Circuit breaker (closed/open/half-open), retry with exponential backoff and jitter, dead letter queue for failed jobs, error coordinator that composes all three.

Quick Start

Installation

go get github.com/abhipray-cpu/morpheus

Top-Level API (Recommended)

The simplest way to use Morpheus. A single import, sensible defaults, all built-in transformers pre-loaded:

package main

import (
    "context"
    "fmt"
    "github.com/abhipray-cpu/morpheus"
)

func main() {
    m := morpheus.New()

    // Single transform
    result, err := m.Transform("uppercase", "hello world")
    fmt.Println(result) // "HELLO WORLD"

    // Chain transforms
    result, err = m.TransformChain([]string{"trim", "uppercase"}, "  hello  ")
    fmt.Println(result) // "HELLO"

    // Detect PII
    ctx := context.Background()
    detections, err := m.DetectPII(ctx, map[string]any{
        "email": "john@example.com",
        "ssn":   "123-45-6789",
    })

    // Mask PII
    masked, err := m.MaskPII(ctx, map[string]any{
        "email": "john@example.com",
    })

    // Rule evaluation
    rules := []string{`
rule: high_value
conditions:
  - field: total
    operator: greater_than
    value: 1000
transformations:
  - field: priority
    transformer: set_value
    params:
      value: high
`}
    data := map[string]any{"total": 1500}
    result, matched, err := m.EvaluateRules(ctx, rules, data)
    fmt.Println(matched) // ["high_value"]

    // Format conversion
    yamlOut, err := m.Convert(ctx, `{"name":"Alice"}`, "json", "yaml")
}

With Options

import (
    "time"
    "github.com/abhipray-cpu/morpheus"
    "github.com/abhipray-cpu/morpheus/pkg/adapter/cache/memory"
)

cache := memory.NewMemoryCache(1000, 5*time.Minute)
m := morpheus.New(
    morpheus.WithCache(cache),
    morpheus.WithCacheTTL(10*time.Minute),
)

Redis Cluster / Sentinel

All Redis adapters (cache, queue, enrichment) accept redis.UniversalClient, which works with standalone, cluster, sentinel, and ring topologies:

import (
    "github.com/redis/go-redis/v9"
    rediscache "github.com/abhipray-cpu/morpheus/pkg/adapter/cache/redis"
)

// Standalone (single node)
client := redis.NewClient(&redis.Options{Addr: "localhost:6379"})
cache := rediscache.NewRedisCacheFromClient(client, 5*time.Minute)

// Cluster (multiple nodes)
cluster := redis.NewClusterClient(&redis.ClusterOptions{
    Addrs: []string{"node1:6379", "node2:6379", "node3:6379"},
})
cache = rediscache.NewRedisCacheFromClient(cluster, 5*time.Minute)

// Sentinel (automatic failover)
failover := redis.NewFailoverClient(&redis.FailoverOptions{
    MasterName:    "mymaster",
    SentinelAddrs: []string{"sentinel1:26379", "sentinel2:26379"},
})
cache = rediscache.NewRedisCacheFromClient(failover, 5*time.Minute)

// Auto-detect via UniversalClient
universal := redis.NewUniversalClient(&redis.UniversalOptions{
    Addrs:      []string{"localhost:6379"},      // 1 addr = standalone, 2+ = cluster
    MasterName: "",                               // set for sentinel mode
})
cache = rediscache.NewRedisCacheFromClient(universal, 5*time.Minute)

Direct Transformer Usage

For finer control, use transformers directly:

import "github.com/abhipray-cpu/morpheus/pkg/domain/entity"

upper := entity.NewUpperCaseTransformer()
ctx := entity.NewTransformContext()
result, err := upper.Transform(ctx, "hello world")
// result = "HELLO WORLD"

Chain Transformers

registry := entity.NewTransformerRegistry()
registry.Register(entity.NewTrimTransformer())
registry.Register(entity.NewUpperCaseTransformer())

chain := entity.NewChainTransformer(registry, []string{"trim", "uppercase"})
result, _ := chain.Transform(entity.NewTransformContext(), "  Hello World  ")
// result: "HELLO WORLD"

Use the Rule Engine

import "github.com/abhipray-cpu/morpheus/pkg/adapter/rule"

cache := memory.NewMemoryCache(1000, 5*time.Minute)
engine := rule.NewRuleEngine(cache, nil)

rules := []string{`
rule: high_value_order
conditions:
  - field: total
    operator: greater_than
    value: 1000
transformations:
  - field: priority
    transformer: set_value
    params:
      value: high
`}

ctx := context.Background()
engine.Compile(ctx, rules)
result, matched, _ := engine.Evaluate(ctx, map[string]any{
    "total": 1500,
})
// result["priority"] = "high", matched = ["high_value_order"]

Documentation

Examples

7 runnable examples covering all features:

make examples           # Run all examples
make run-example E=01   # Run a specific example

Architecture

Morpheus uses hexagonal architecture with three layers:

Domain Layer (entities, services, value objects)
  Transformer interface + 50+ implementations
  Pipeline, Stage, Job entities
  TransformationService, ValidationService
  FormatConverter, StreamParsers

Port Layer (interfaces)
  TransformerPort, PipelineExecutorPort
  RuleEnginePort, PrivacyPort
  CachePort, MetricsPort

Adapter Layer (implementations)
  Rule Engine, Pipeline Orchestrator
  PII Detector/Masker/Tokenizer
  Memory/Redis/Multi-level Cache
  Circuit Breaker, Retry, DLQ

See Architecture Documentation for full diagrams.

Dependencies

Core transformers have zero external dependencies. Adapters that connect to external systems pull in:

Development

make setup              # Install dependencies
make test               # Run all tests
make test-race          # Race detector
make test-coverage      # Coverage report
make lint               # Lint
make validate           # All checks
make fmt                # Format code

Test Suite

• 1300+ tests across unit, integration, and e2e
• Race detector clean (on non-Redis tests)

Contributing

See CONTRIBUTING.md for guidelines.

git checkout -b feature/my-feature
# make your changes
make validate
git commit -m "feat: add my feature"
git push origin feature/my-feature
# open a PR

License

MIT. See LICENSE.