Skip to content

Writing Modules

Jump to bottom
TheMinecraftGuyGuru edited this page Jun 8, 2026 · 5 revisions

Writing Modules

This guide covers building a MuxCore module from scratch against the v1 fabric-only core.

Quick Start

A module is a Go package that:

  1. Implements contracts.Module (lifecycle)
  2. Calls contracts.Register() in init() (auto-discovery)
  3. Implements one or more domain contract interfaces (from contracts-* repos)
  4. Has a muxcore.json at its repo root (marketplace metadata)
my-module/
  go.mod           <- depends on core + contract repos
  module.go        <- your implementation
  muxcore.json     <- marketplace metadata

Minimal Module

package mymodule

import (
    "context"
    "github.com/Muxcore-Media/core/pkg/contracts"
)

func init() {
    contracts.Register(func(deps contracts.ModuleDeps) contracts.Module {
        return NewModule(deps.EventBus, deps.Registry)
    })
}

type Module struct {
    bus contracts.EventBus
    reg contracts.ServiceRegistry
}

func NewModule(bus contracts.EventBus, reg contracts.ServiceRegistry) *Module {
    return &Module{bus: bus, reg: reg}
}

func (m *Module) Info() contracts.ModuleInfo {
    return contracts.ModuleInfo{
        ID:      "my-module",
        Name:    "My Module",
        Version: "1.0.0",
        Kinds:   []string{"provider"},
    }
}

func (m *Module) Init(ctx context.Context) error  { return nil }
func (m *Module) Start(ctx context.Context) error { return nil }
func (m *Module) Stop(ctx context.Context) error  { return nil }
func (m *Module) Health(ctx context.Context) error { return nil }

ModuleDeps — What Core Gives You

Every module receives contracts.ModuleDeps with eight fabric services:

Field Interface Use
Registry ServiceRegistry Discover other modules at runtime
EventBus EventBus Publish and subscribe to events
Routes RouteRegistrar Register HTTP handlers
Cluster Cluster Cluster membership (nil if standalone)
Storage StorageOrchestrator Object storage via IDs, not paths
WorkerPool WorkerPool Distributed task scheduling
Audit AuditLogger System-wide audit logging
Mesh ModuleMeshClient Cross-module gRPC calls

Services NOT in ModuleDeps — discover at runtime via Registry.FindByCapability():

// Discover secrets provider
entries := deps.Registry.FindByCapability("secrets")
if len(entries) > 0 {
    secrets = entries[0].Module.(contracts.SecretsProvider)
}

// Discover cache layer (for storage read-through caching)
entries = deps.Registry.FindByCapability("cache.local")
if len(entries) > 0 {
    cache = entries[0].Module.(contracts.CacheLayer)
}

How to Implement Each Module Kind

Downloader

Import github.com/Muxcore-Media/contracts-downloader. Implement dl.Downloader. Kind: "downloader".

import dl "github.com/Muxcore-Media/contracts-downloader"

func (m *Module) Add(ctx context.Context, task dl.DownloadTask) (string, error) { ... }
// task.Extra carries module-specific context (HTTP URLs, streaming params, etc.)

Indexer

Import github.com/Muxcore-Media/contracts-indexer. Implement indexer.Indexer. Kind: "indexer".

Playback

Import github.com/Muxcore-Media/contracts-playback. Implement playback.Playback and/or playback.StreamSource. Kind: "playback".

Media Manager

Import github.com/Muxcore-Media/contracts-media. Implement media.MediaLibrary. Kind: "media_manager".

import media "github.com/Muxcore-Media/contracts-media"

func (m *Module) List(ctx context.Context, mediaType media.MediaType, filter *media.MediaFilter, offset, limit int) ([]media.MediaObject, error) { ... }
// filter.Extra carries media-type-specific filters (artist, album, BPM, key, etc.)

Storage Provider

Implement contracts.StorageProvider (in core). Optionally implement Streamable, Seekable, Watchable, AtomicMovable, Hardlinkable. Kind: "storage".

Scheduler

Implement contracts.Scheduler (in core). Kind: "scheduler". SchedulerTask includes Extra map[string]any.

Auth Provider

Implement contracts.AuthProvider (in core). Kind: "auth". Core wires the first auth module into API middleware automatically.

UI / API Module

Register HTTP handlers via deps.Routes. Kinds: "ui", "api".

Infrastructure Modules

Interface Capability Discovery
SecretsProvider "secrets" FindByCapability("secrets")
DatabaseProvider "database" FindByCapability("database")
CacheProvider "cache" FindByCapability("cache")
CacheLayer "cache.local" FindByCapability("cache.local")
RateLimiterProvider — Core picks up automatically
HealthMonitor — Core picks up automatically
AuditLogger — Injected via ModuleDeps.Audit
Cluster — Injected via ModuleDeps.Cluster
TracingProvider "tracing" FindByCapability("tracing")
CircuitBreaker "circuitbreaker" FindByCapability("circuitbreaker")
ConfigWatcher "config.watcher" FindByCapability("config.watcher")
DeadLetterProvider "deadletter" FindByCapability("deadletter")
CallPolicyProvider "call.policy" FindByCapability("call.policy")
IdentityProvider "identity" FindByCapability("identity")
StructuredLogger "logging" FindByCapability("logging")
RetryProvider "retry" FindByCapability("retry")
IdempotencyProvider "idempotency" FindByCapability("idempotency")
FeatureFlagProvider "feature.flags" FindByCapability("feature.flags")
SerializationProvider "serialization" FindByCapability("serialization")
EncryptionProvider "encryption" FindByCapability("encryption")
DistributedLockProvider "distributed.lock" FindByCapability("distributed.lock")
DataRedactionProvider "data.redaction" FindByCapability("data.redaction")
EventStore "event.store" FindByCapability("event.store")
Authorizer — Core picks up automatically
WorkerPool — Injected via ModuleDeps.WorkerPool

Module Communication

Discovering Other Modules

// Broad: find all downloaders
entries := deps.Registry.FindByKind("downloader")

// Fine-grained: find by capability
entries := deps.Registry.FindByCapability("playback.jellyfin")

// Direct: resolve by ID
entry, err := deps.Registry.Resolve("downloader-qbittorrent")

// Type-assert to the domain contract
import dl "github.com/Muxcore-Media/contracts-downloader"
for _, entry := range entries {
    if d, ok := entry.Module.(dl.Downloader); ok {
        d.Add(ctx, task)
    }
}

Publishing Events

Event types and payloads are defined in contract repos:

import dl "github.com/Muxcore-Media/contracts-downloader"

payload, _ := json.Marshal(dl.DownloadCompletedPayload{
    DownloadID: id,
    Title:      title,
})

deps.EventBus.Publish(ctx, contracts.Event{
    Type:    dl.EventDownloadCompleted,
    Source:  "my-downloader",
    Payload: payload,
})

Subscribing to Events

import dl "github.com/Muxcore-Media/contracts-downloader"

deps.EventBus.Subscribe(ctx, dl.EventDownloadCompleted,
    func(ctx context.Context, event contracts.Event) error {
        var payload dl.DownloadCompletedPayload
        json.Unmarshal(event.Payload, &payload)
        return nil
    })

Cross-Module gRPC Calls

result, err := deps.Mesh.Call(ctx, "transcoder-ffmpeg", "Status", jsonPayload)

With Call Policy

If a CallPolicyProvider is registered, the mesh client enforces access control. Set your module ID in the context before calling:

ctx = contracts.WithCallerID(ctx, "my-module")
result, err := deps.Mesh.Call(ctx, "target-module", "method", payload)

With Circuit Breaker

Wrap cross-module calls in a circuit breaker to prevent cascading failures:

breaker := discoveredContracts.CircuitBreaker
err := breaker.Execute(ctx, "target-module", func(ctx context.Context) error {
    _, err := deps.Mesh.Call(ctx, "target-module", "method", payload)
    return err
})
if errors.Is(err, contracts.ErrCircuitOpen) {
    // circuit is open, handle gracefully
}

Module Metadata (muxcore.json)

{
  "name": "My Downloader",
  "description": "A torrent downloader",
  "version": "1.0.0",
  "kind": "downloader",
  "capabilities": ["downloader.torrent"],
  "contracts": [
    {"repo": "github.com/Muxcore-Media/contracts-downloader", "version": "v1.0.0", "interface": "Downloader"}
  ],
  "expects": [
    {"repo": "github.com/Muxcore-Media/contracts-secrets", "version": "v1.0.0", "interface": "SecretsProvider"}
  ],
  "dependencies": [],
  "homepage": "https://github.com/you/my-module"
}
Field Required Notes
name Yes Display name
description Yes Short description
version Yes Semantic version
kind Yes Primary kind string (e.g., "downloader")
capabilities Yes Array of capability strings
contracts Yes Contracts this module implements
expects No Contracts this module needs from others
dependencies No Module IDs this depends on
homepage No URL to repo

Publishing

  1. Push module to a public GitHub repo
  2. Add repo URL to a marketplace catalog.json
  3. For the official marketplace, submit a PR to github.com/Muxcore-Media/marketplace-catalog

Using Tracing

Create spans around operations for distributed tracing:

tracing, _ := deps.Registry.FindByCapability("tracing")
if len(tracing) > 0 {
    tp := tracing[0].Module.(contracts.TracingProvider)
    ctx, span := tp.StartSpan(ctx, "indexer.search")
    defer span.End()
    span.SetAttribute("query", query)
    // ... do work ...
    if err != nil {
        span.SetStatus(contracts.SpanStatusError, err.Error())
    }
}

Using ConfigWatcher

React to runtime changes without polling:

watcher, _ := deps.Registry.FindByCapability("config.watcher")
if len(watcher) > 0 {
    cw := watcher[0].Module.(contracts.ConfigWatcher)
    cancel, _ := cw.OnChange(ctx, "secrets", func() {
        // Re-discover secrets provider
        entries := deps.Registry.FindByCapability("secrets")
        if len(entries) > 0 {
            secrets = entries[0].Module.(contracts.SecretsProvider)
        }
    })
    defer cancel()
}

Using DeadLetterProvider

Handle critical event failures gracefully:

dl, _ := deps.Registry.FindByCapability("deadletter")
deps.EventBus.Subscribe(ctx, "workflow.step.completed", func(ctx context.Context, event contracts.Event) error {
    if err := processEvent(event); err != nil {
        if len(dl) > 0 {
            dl[0].Module.(contracts.DeadLetterProvider).Store(ctx, event, "my-handler", err)
        }
        return err
    }
    return nil
})

Using IdentityProvider

Extract caller identity for authorization checks:

identity, _ := deps.Registry.FindByCapability("identity")
if len(identity) > 0 {
    idp := identity[0].Module.(contracts.IdentityProvider)
    who, err := idp.ExtractIdentity(ctx)
    if err != nil {
        // invalid token, reject
        return contracts.ErrUnauthorized
    }
    if who == nil {
        // unauthenticated — apply anonymous policy
    } else {
        // pass to Authorizer
        deps.Authorizer.Authorize(ctx, who.ID, "storage.write")
    }
}

Using StructuredLogger

Log with structured fields instead of raw prints:

logger, _ := deps.Registry.FindByCapability("logging")
if len(logger) > 0 {
    log := logger[0].Module.(contracts.StructuredLogger)
    log.Info(ctx, "download started", map[string]any{
        "task_id":    task.ID,
        "magnet_uri": task.MagnetURI,
    })
    if err != nil {
        log.Error(ctx, "download failed", map[string]any{
            "task_id": task.ID,
            "error":   err.Error(),
        })
    }
}

Using RetryProvider

Retry transient failures with exponential backoff:

retry, _ := deps.Registry.FindByCapability("retry")
if len(retry) > 0 {
    rp := retry[0].Module.(contracts.RetryProvider)
    err := rp.Execute(ctx, contracts.RetryPolicy{
        MaxAttempts:   5,
        InitialDelay:  100 * time.Millisecond,
        BackoffFactor: 2.0,
        Jitter:        true,
    }, func(ctx context.Context) error {
        return deps.Mesh.Call(ctx, "upstream", "Fetch", payload)
    })
}
// If no RetryProvider registered, Execute runs fn once — safe either way.

Using IdempotencyProvider

Prevent duplicate processing of the same request:

idem, _ := deps.Registry.FindByCapability("idempotency")
if len(idem) > 0 {
    ip := idem[0].Module.(contracts.IdempotencyProvider)
    isNew, err := ip.Store(ctx, requestID, cachedResult, 24*time.Hour)
    if err != nil {
        return err
    }
    if !isNew {
        // Already processed — return cached result
        cached, _ := ip.Result(ctx, requestID)
        return cached, nil
    }
}
// Process the request (first time only)...

Using FeatureFlagProvider

Gate new features behind flags:

flags, _ := deps.Registry.FindByCapability("feature.flags")
enabled := false
if len(flags) > 0 {
    ff := flags[0].Module.(contracts.FeatureFlagProvider)
    enabled = ff.IsEnabled(ctx, "new-recommendation-engine", false)
}
if enabled {
    // run new code path
}
// For A/B testing:
variant := "control"
if len(flags) > 0 {
    variant = flags[0].Module.(contracts.FeatureFlagProvider).GetVariant(ctx, "ui-layout", "control")
}

Using SerializationProvider

Marshal/unmarshal without hardcoding format:

ser, _ := deps.Registry.FindByCapability("serialization")
if len(ser) > 0 {
    sp := ser[0].Module.(contracts.SerializationProvider)
    // Negotiate: pick first format both sides support
    data, err := sp.Marshal("application/json", myStruct)
    // Or let the provider handle format conversion:
    var result MyType
    sp.Unmarshal("application/x-protobuf", wireBytes, &result)
}

Using EncryptionProvider

Encrypt sensitive data before storage:

enc, _ := deps.Registry.FindByCapability("encryption")
if len(enc) > 0 {
    ep := enc[0].Module.(contracts.EncryptionProvider)
    ciphertext, err := ep.Encrypt(ctx, sensitiveData)
    if err != nil {
        return err
    }
    // Store ciphertext (self-describing, safe to decrypt later even after rotation)
    deps.Storage.Put(ctx, "sensitive/key", ciphertext)

    // Later:
    stored, _ := deps.Storage.Get(ctx, "sensitive/key")
    plaintext, err := ep.Decrypt(ctx, stored)

    // Rotate keys on schedule:
    ep.RotateKey(ctx)
}

Using DistributedLockProvider

Coordinate across modules without relying on cache locks:

lock, _ := deps.Registry.FindByCapability("distributed.lock")
if len(lock) > 0 {
    lp := lock[0].Module.(contracts.DistributedLockProvider)
    handle, err := lp.Acquire(ctx, "migration-leader", 30*time.Second)
    if errors.Is(err, contracts.ErrLockHeld) {
        // Another instance is running the migration, skip
        return nil
    }
    if err != nil {
        return err
    }
    defer handle.Unlock(ctx)

    // For long operations, extend the lease:
    handle.Renew(ctx, 30*time.Second)

    // ... critical section ...
}
// If no DistributedLockProvider, always succeeds — single-node mode.

Using DataRedactionProvider

Redact sensitive fields before logging or audit:

redact, _ := deps.Registry.FindByCapability("data.redaction")
data := map[string]any{
    "user_id":  "abc123",
    "email":    "user@example.com",
    "action":   "login",
    "token":    "sk-abc123secret",
}

if len(redact) > 0 {
    rp := redact[0].Module.(contracts.DataRedactionProvider)
    safe, _ := rp.Redact(ctx, data, []string{"email", "token"})
    // safe = {"user_id": "abc123", "email": "***REDACTED***", "action": "login", "token": "***REDACTED***"}
    deps.Audit.Log(ctx, contracts.AuditEntry{
        Action: "user.login",
        Data:   safe,
    })
}
// If no DataRedactionProvider, data passes through unchanged — log safely.

Using EventStore

Append to and replay event streams for event sourcing:

store, _ := deps.Registry.FindByCapability("event.store")
if len(store) > 0 {
    es := store[0].Module.(contracts.EventStore)

    // Append events atomically:
    seq, err := es.Append(ctx, "user-abc123", []contracts.Event{
        {Type: "user.created", Source: "auth-module", Payload: userJSON},
        {Type: "user.verified", Source: "auth-module", Payload: verifiedJSON},
    })
    // seq = 1 (first sequence number in the batch)

    // Rebuild state by replaying from the beginning:
    events, _ := es.Read(ctx, "user-abc123", 1, 0) // from seq 1, no limit
    for _, entry := range events {
        switch entry.Event.Type {
        case "user.created":
            applyUserCreated(entry.Event.Payload)
        case "user.verified":
            applyUserVerified(entry.Event.Payload)
        }
    }

    // Subscribe to live events from the last seen sequence:
    ch, _ := es.Subscribe(ctx, "user-abc123", events[len(events)-1].Sequence+1)
    for entry := range ch {
        handleEvent(entry)
    }
}

Checklist

  • go.mod depends on github.com/Muxcore-Media/core plus contract repos you implement
  • contracts.Register() called in init()
  • Implements contracts.Module (Info, Init, Start, Stop, Health)
  • Implements the domain contract from the appropriate contracts-* repo
  • ContractDeclarations in Info().Contracts for compatibility discovery
  • muxcore.json at repo root with contracts and expects fields
  • Module discovers services it needs via Registry.FindByCapability()

Wiki Navigation

Clone this wiki locally