From b9571d55f3f0595396d730c38118debac33c8f23 Mon Sep 17 00:00:00 2001 From: Pierre Date: Sun, 11 Jan 2026 17:07:52 +0100 Subject: [PATCH] initial commit --- .github/workflows/docker-publish.yml | 59 +++ .gitignore | 22 + Dockerfile | 64 +++ Makefile | 109 ++++ README.md | 410 +++++++++++++++ cmd/meta-core/main.go | 111 +++++ go.mod | 13 + go.sum | 8 + internal/api/handlers.go | 720 +++++++++++++++++++++++++++ internal/api/server.go | 146 ++++++ internal/config/config.go | 85 ++++ internal/discovery/service.go | 258 ++++++++++ internal/leader/election.go | 430 ++++++++++++++++ internal/leader/redis.go | 179 +++++++ internal/storage/client.go | 549 ++++++++++++++++++++ test/api_test.go | 174 +++++++ test/config_test.go | 91 ++++ test/election_test.go | 144 ++++++ 18 files changed, 3572 insertions(+) create mode 100644 .github/workflows/docker-publish.yml create mode 100644 .gitignore create mode 100644 Dockerfile create mode 100644 Makefile create mode 100644 README.md create mode 100644 cmd/meta-core/main.go create mode 100644 go.mod create mode 100644 go.sum create mode 100644 internal/api/handlers.go create mode 100644 internal/api/server.go create mode 100644 internal/config/config.go create mode 100644 internal/discovery/service.go create mode 100644 internal/leader/election.go create mode 100644 internal/leader/redis.go create mode 100644 internal/storage/client.go create mode 100644 test/api_test.go create mode 100644 test/config_test.go create mode 100644 test/election_test.go diff --git a/.github/workflows/docker-publish.yml b/.github/workflows/docker-publish.yml new file mode 100644 index 0000000..6b755d0 --- /dev/null +++ b/.github/workflows/docker-publish.yml @@ -0,0 +1,59 @@ +name: Build and Push Docker Image + +on: + push: + tags: + - 'v*' + workflow_dispatch: + inputs: + tag: + description: 'Tag for the image (optional, defaults to sha)' + required: false + type: string + +env: + REGISTRY: ghcr.io + IMAGE_NAME: ${{ github.repository }} + +jobs: + build: + runs-on: ubuntu-latest + permissions: + contents: read + packages: write + + steps: + - name: Checkout repository + uses: actions/checkout@v4 + + - name: Set up Docker Buildx + uses: docker/setup-buildx-action@v3 + + - name: Log in to Container Registry + uses: docker/login-action@v3 + with: + registry: ${{ env.REGISTRY }} + username: ${{ github.actor }} + password: ${{ secrets.GITHUB_TOKEN }} + + - name: Extract metadata (tags, labels) + id: meta + uses: docker/metadata-action@v5 + with: + images: ${{ env.REGISTRY }}/${{ env.IMAGE_NAME }} + tags: | + type=semver,pattern={{version}} + type=semver,pattern={{major}}.{{minor}} + type=sha,prefix= + type=raw,value=latest,enable={{is_default_branch}} + type=raw,value=${{ inputs.tag }},enable=${{ inputs.tag != '' }} + + - name: Build and push Docker image + uses: docker/build-push-action@v5 + with: + context: . + push: true + tags: ${{ steps.meta.outputs.tags }} + labels: ${{ steps.meta.outputs.labels }} + cache-from: type=gha + cache-to: type=gha,mode=max diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..203bd83 --- /dev/null +++ b/.gitignore @@ -0,0 +1,22 @@ +# Binaries +bin/ +*.exe +*.exe~ +*.dll +*.so +*.dylib + +# Test binary, build with `go test -c` +*.test + +# Output of the go coverage tool +*.out + +# Go workspace file +go.work + +# IDE +.idea/ +.vscode/ +*.swp +*.swo diff --git a/Dockerfile b/Dockerfile new file mode 100644 index 0000000..5ed95cf --- /dev/null +++ b/Dockerfile @@ -0,0 +1,64 @@ +# Build stage +FROM golang:1.21-alpine AS builder + +# Install build dependencies +RUN apk add --no-cache git + +WORKDIR /build + +# Copy go mod files first for caching +COPY go.mod go.sum ./ +RUN go mod download + +# Copy source code +COPY . . + +# Build the binary +RUN CGO_ENABLED=0 GOOS=linux go build -ldflags="-s -w" -o meta-core ./cmd/meta-core + +# Runtime stage +FROM alpine:3.19 + +# Container registry metadata +LABEL org.opencontainers.image.source=https://github.com/worph/meta-core +LABEL org.opencontainers.image.description="MetaMesh meta-core sidecar - leader election, Redis management, HTTP API" +LABEL org.opencontainers.image.licenses=MIT + +# Install runtime dependencies +RUN apk add --no-cache \ + redis \ + bash \ + ca-certificates + +# Create non-root user +RUN addgroup -S metacore && adduser -S metacore -G metacore + +# Copy binary from builder +COPY --from=builder /build/meta-core /usr/local/bin/meta-core + +# Set permissions +RUN chmod +x /usr/local/bin/meta-core + +# Create directories +RUN mkdir -p /meta-core/locks /meta-core/db/redis /meta-core/services && \ + chown -R metacore:metacore /meta-core + +# Switch to non-root user +USER metacore + +# Environment defaults +ENV META_CORE_PATH=/meta-core \ + FILES_PATH=/files \ + SERVICE_NAME=meta-core \ + META_CORE_HTTP_PORT=9000 \ + REDIS_PORT=6379 + +# Expose HTTP API port +EXPOSE 9000 + +# Health check +HEALTHCHECK --interval=10s --timeout=5s --start-period=5s --retries=3 \ + CMD wget -q --spider http://localhost:9000/health || exit 1 + +# Run the binary +ENTRYPOINT ["/usr/local/bin/meta-core"] diff --git a/Makefile b/Makefile new file mode 100644 index 0000000..d39848b --- /dev/null +++ b/Makefile @@ -0,0 +1,109 @@ +.PHONY: build test clean docker lint run help + +# Variables +BINARY := meta-core +VERSION ?= 1.0.0 +BUILD_DIR := bin +DOCKER_IMAGE := meta-core + +# Go build flags +LDFLAGS := -ldflags="-s -w -X main.Version=$(VERSION)" +BUILD_FLAGS := CGO_ENABLED=0 + +# Default target +all: build + +# Build the binary +build: + @echo "Building $(BINARY)..." + @mkdir -p $(BUILD_DIR) + $(BUILD_FLAGS) go build $(LDFLAGS) -o $(BUILD_DIR)/$(BINARY) ./cmd/meta-core + +# Build for Linux (cross-compile) +build-linux: + @echo "Building $(BINARY) for Linux..." + @mkdir -p $(BUILD_DIR) + $(BUILD_FLAGS) GOOS=linux GOARCH=amd64 go build $(LDFLAGS) -o $(BUILD_DIR)/$(BINARY)-linux-amd64 ./cmd/meta-core + +# Run tests +test: + @echo "Running tests..." + go test -v ./... + +# Run tests with coverage +test-cover: + @echo "Running tests with coverage..." + go test -v -coverprofile=coverage.out ./... + go tool cover -html=coverage.out -o coverage.html + @echo "Coverage report: coverage.html" + +# Run linter +lint: + @echo "Running linter..." + @if command -v golangci-lint >/dev/null 2>&1; then \ + golangci-lint run; \ + else \ + echo "golangci-lint not installed, running go vet instead"; \ + go vet ./...; \ + fi + +# Format code +fmt: + @echo "Formatting code..." + go fmt ./... + +# Tidy dependencies +tidy: + @echo "Tidying dependencies..." + go mod tidy + +# Clean build artifacts +clean: + @echo "Cleaning..." + rm -rf $(BUILD_DIR) + rm -f coverage.out coverage.html + +# Build Docker image +docker: + @echo "Building Docker image..." + docker build -t $(DOCKER_IMAGE):$(VERSION) . + docker tag $(DOCKER_IMAGE):$(VERSION) $(DOCKER_IMAGE):latest + +# Run locally (for development) +run: build + @echo "Running $(BINARY)..." + META_CORE_PATH=/tmp/meta-core \ + FILES_PATH=/tmp/files \ + SERVICE_NAME=meta-core-dev \ + META_CORE_HTTP_PORT=9000 \ + ./$(BUILD_DIR)/$(BINARY) + +# Install to system +install: build + @echo "Installing $(BINARY) to /usr/local/bin..." + sudo cp $(BUILD_DIR)/$(BINARY) /usr/local/bin/ + +# Check if Go is installed +check: + @echo "Go version:" + @go version + @echo "" + @echo "Dependencies:" + @go list -m all + +# Help +help: + @echo "Available targets:" + @echo " build - Build the binary" + @echo " build-linux - Build for Linux (cross-compile)" + @echo " test - Run tests" + @echo " test-cover - Run tests with coverage" + @echo " lint - Run linter" + @echo " fmt - Format code" + @echo " tidy - Tidy dependencies" + @echo " clean - Clean build artifacts" + @echo " docker - Build Docker image" + @echo " run - Build and run locally" + @echo " install - Install to /usr/local/bin" + @echo " check - Check Go installation and dependencies" + @echo " help - Show this help" diff --git a/README.md b/README.md new file mode 100644 index 0000000..f0280a8 --- /dev/null +++ b/README.md @@ -0,0 +1,410 @@ +# meta-core + +A Go-based sidecar binary that provides unified leader election, Redis management, and HTTP API for MetaMesh services. + +## Overview + +meta-core runs alongside each MetaMesh service (meta-sort, meta-fuse, meta-stremio) as a sidecar process managed by supervisord. It provides the foundational infrastructure that enables MetaMesh's distributed architecture without requiring external coordination services like etcd or Consul. + +### Key Capabilities + +| Feature | Description | +|---------|-------------| +| **Leader Election** | POSIX flock-based distributed consensus on shared filesystem | +| **Redis Management** | Leader spawns Redis with AOF+RDB persistence, auto-restart on crash | +| **HTTP API** | Language-agnostic REST interface for metadata and service discovery | +| **Service Discovery** | File-based registry with heartbeats and stale detection | +| **Metadata Storage** | Flat key-value schema with connection pooling and batch operations | + +### Design Characteristics + +- **No external dependencies** - Uses filesystem locks (works on NFS/CIFS shared volumes) +- **Automatic failover** - Lock releases on process death, followers detect and re-elect +- **Thread-safe** - RWMutex protection on all shared state +- **Graceful shutdown** - Signal handling with reverse startup order +- **Static binary** - CGO disabled, single binary deployment + +## Architecture + +``` +┌─────────────────────────────────────────────────────────────┐ +│ Container (meta-sort / meta-fuse / meta-stremio) │ +│ │ +│ ┌──────────────┐ HTTP API ┌──────────────────┐ │ +│ │ Main Service │ ◄────────────────► │ meta-core │ │ +│ │ (TS/Python) │ localhost:9000 │ (Go binary) │ │ +│ └──────────────┘ └────────┬─────────┘ │ +│ │ │ +└───────────────────────────────────────────────┼─────────────┘ + │ + ┌───────────────────────┴───────────────┐ + │ /meta-core (shared volume) │ + │ ├── locks/kv-leader.lock │ + │ ├── locks/kv-leader.info │ + │ ├── db/redis/ │ + │ └── services/*.json │ + └───────────────────────────────────────┘ +``` + +## Quick Start + +### Build + +```bash +# Build binary +make build + +# Build Docker image +make docker + +# Run tests +make test +``` + +### Run + +```bash +# With environment variables +META_CORE_PATH=/meta-core \ +FILES_PATH=/files \ +SERVICE_NAME=meta-sort \ +META_CORE_HTTP_PORT=9000 \ +./bin/meta-core +``` + +### Docker + +```bash +docker build -t meta-core . +docker run -v meta-core:/meta-core -v files:/files meta-core +``` + +## Configuration + +| Environment Variable | Default | Description | +|---------------------|---------|-------------| +| `META_CORE_PATH` | `/meta-core` | Shared volume for locks/db | +| `FILES_PATH` | `/files` | Shared volume for media files | +| `SERVICE_NAME` | `meta-core` | Service identifier | +| `SERVICE_VERSION` | `1.0.0` | Service version | +| `API_PORT` | `8180` | Main service HTTP port | +| `BASE_URL` | - | Stable service URL | +| `REDIS_PORT` | `6379` | Redis port (leader only) | +| `META_CORE_HTTP_PORT` | `9000` | HTTP API port | +| `META_CORE_HTTP_HOST` | `127.0.0.1` | HTTP API bind address | +| `HEALTH_CHECK_INTERVAL_MS` | `5000` | Health check interval | +| `HEARTBEAT_INTERVAL_MS` | `30000` | Service heartbeat interval | +| `STALE_THRESHOLD_MS` | `60000` | Stale service threshold | + +## API Reference + +### Health & Status + +```bash +# Health check +curl http://localhost:9000/health +# {"status":"ok","role":"leader","redis":true,"timestamp":"..."} + +# Detailed status +curl http://localhost:9000/status +# {"status":"ok","role":"leader","serviceName":"meta-core","version":"1.0.0",...} + +# Current leader +curl http://localhost:9000/leader +# {"host":"meta-sort-dev","api":"redis://10.0.1.50:6379","http":"http://10.0.1.50:8180",...} + +# This instance's role +curl http://localhost:9000/role +# {"role":"leader"} +``` + +### Metadata Operations + +```bash +# List all file hashes +curl http://localhost:9000/meta +# {"hashIds":["midhash256:abc123",...],"count":42} + +# Get metadata for a file +curl http://localhost:9000/meta/{hash} +# {"hashId":"midhash256:abc123","metadata":{"title":"Movie","year":"2024",...}} + +# Update metadata +curl -X PUT http://localhost:9000/meta/{hash} \ + -H "Content-Type: application/json" \ + -d '{"title":"New Title"}' + +# Delete metadata +curl -X DELETE http://localhost:9000/meta/{hash} +``` + +### Data Operations + +```bash +# Get file path +curl http://localhost:9000/data/{hash}/path +# {"hashId":"midhash256:abc123","path":"/files/movies/Movie.mkv","exists":true} + +# Check if file exists +curl -I http://localhost:9000/data/{hash} +# HTTP/1.1 200 OK (or 404) +``` + +### File Operations (by CID) + +```bash +# Serve file by CID (looks up poster/backdrop CIDs in metadata) +curl http://localhost:9000/file/{cid} --output poster.jpg +# Returns raw file bytes with appropriate Content-Type header + +# Example: fetch a poster image +curl -v http://localhost:9000/file/bafkreih5aznjvttude6c3wbvqeebb6rlx5wkbzyppv7garber7ndsuxku4 +# Content-Type: image/jpeg +# [binary image data] +``` + +The `/file/{cid}` endpoint searches all file metadata for matching `poster` or `backdrop` CIDs and serves the corresponding file from disk. Supports range requests for partial content retrieval. + +### Service Discovery + +```bash +# List all services +curl http://localhost:9000/services +# {"services":[{"name":"meta-sort",...}],"count":3} + +# Get specific service +curl http://localhost:9000/services/meta-sort +# {"name":"meta-sort","version":"2.0.0","api":"http://...","status":"running",...} +``` + +## Leader Election + +meta-core uses POSIX file locking (flock) for distributed leader election: + +1. Each instance tries to acquire exclusive lock on `/meta-core/locks/kv-leader.lock` +2. Winner becomes **leader**, spawns Redis, writes info to `/meta-core/locks/kv-leader.info` +3. Losers become **followers**, read leader info, connect to leader's Redis +4. Lock automatically releases when process dies (no stale locks) + +### Election Flow + +``` +Instance 1 Instance 2 Instance 3 + │ │ │ + ├── Try flock ──────────┼───────────────────────┤ + │ │ │ + [WIN] [LOSE] [LOSE] + LEADER FOLLOWER FOLLOWER + │ │ │ + Start Redis Read leader.info Read leader.info + │ │ │ + Write leader.info Connect to Redis Connect to Redis + │ │ │ + [Health loop 5s] [Health loop 5s] [Health loop 5s] + - Update timestamp - Re-read leader - Re-read leader + - Check Redis alive - Detect changes - Detect changes +``` + +### Role Callbacks + +The election system provides callbacks for role transitions: + +```go +election.OnBecomeLeader(func(info LeaderLockInfo) { + // Start Redis, initialize resources +}) +election.OnBecomeFollower(func(info LeaderLockInfo) { + // Connect to leader's Redis +}) +election.OnLeaderLost(func() { + // Handle leader failure, attempt re-election +}) +``` + +### Lock Info Format + +```json +{ + "host": "meta-sort-dev", + "api": "redis://10.0.1.50:6379", + "http": "http://10.0.1.50:8180", + "baseUrl": "http://localhost:8180", + "timestamp": 1704067200000, + "pid": 12345 +} +``` + +## Redis Management + +The leader is responsible for spawning and managing the Redis server. + +### Redis Configuration + +```bash +redis-server --port 6379 --bind 0.0.0.0 \ + --dir /meta-core/db/redis \ + --appendonly yes --appendfilename appendonly.aof \ + --dbfilename dump.rdb --save 60 1 \ + --loglevel warning +``` + +### Persistence Strategy + +| Method | Configuration | Purpose | +|--------|--------------|---------| +| **AOF** | `appendonly yes` | Write-ahead log for durability | +| **RDB** | `save 60 1` | Snapshot after 60s if ≥1 key changed | + +### Failure Handling + +- **Redis crash**: Auto-detected via health check, automatically restarted +- **Leader crash**: Lock released, followers detect via stale timestamp, re-election occurs +- **Graceful shutdown**: SIGTERM sent to Redis (10s timeout), then SIGKILL if needed + +## Metadata Storage + +meta-core uses a flat key-value schema in Redis for storing file metadata. + +### Storage Format + +``` +Key: /file/{hashId}/{property} +Value: string + +Examples: +/file/midhash256:abc123/title → "Movie Title" +/file/midhash256:abc123/year → "2024" +/file/midhash256:abc123/genres → "action,drama" +/file/midhash256:abc123/filePath → "movies/Movie.mkv" +``` + +### Storage Client Features + +| Feature | Details | +|---------|---------| +| Connection Pool | 10 connections | +| Timeouts | 5s dial, 30s read/write | +| Batch Operations | Pipelined writes for performance | +| Key Scanning | SCAN-based iteration (non-blocking) | +| CID Lookup | Find file paths by poster/backdrop CID | + +## Service Discovery + +Services register by writing JSON to `/meta-core/services/{name}.json`. + +### Registration Format + +```json +{ + "name": "meta-sort", + "version": "2.0.0", + "api": "http://10.0.1.50:8180", + "status": "running", + "pid": 12345, + "hostname": "meta-sort-dev", + "startedAt": "2024-01-01T00:00:00Z", + "lastHeartbeat": "2024-01-01T00:01:00Z", + "capabilities": ["meta-core"], + "endpoints": { + "health": "http://10.0.1.50:9000/health", + "meta": "http://10.0.1.50:9000/meta" + } +} +``` + +### Heartbeat & Stale Detection + +| Parameter | Default | Purpose | +|-----------|---------|---------| +| `HEARTBEAT_INTERVAL_MS` | 30000 | How often services update `lastHeartbeat` | +| `STALE_THRESHOLD_MS` | 60000 | Mark service "stale" if heartbeat older than this | + +Services are automatically marked as `stale` when discovered if their `lastHeartbeat` exceeds the threshold. On graceful shutdown, services remove their registration file. + +## Development + +```bash +# Format code +make fmt + +# Run linter +make lint + +# Run tests with coverage +make test-cover + +# Clean build artifacts +make clean +``` + +## Internal Packages + +| Package | Purpose | +|---------|---------| +| `cmd/meta-core` | Main entry point, startup/shutdown orchestration | +| `internal/config` | Environment-based configuration with defaults | +| `internal/leader` | Election system (`election.go`) and Redis manager (`redis.go`) | +| `internal/storage` | Redis client wrapper with flat key-value operations | +| `internal/discovery` | Service registration, heartbeat loop, discovery | +| `internal/api` | HTTP server, router, and all endpoint handlers | + +### Startup Sequence + +``` +1. Load configuration from environment +2. Create storage client (Redis wrapper) +3. Initialize leader election +4. Register role transition callbacks +5. Start election (acquire lock or become follower) +6. Start service discovery (register + heartbeat loop) +7. Start HTTP API server +8. Wait for SIGINT/SIGTERM +9. Shutdown in reverse order +``` + +## Integration + +### As Sidecar in Docker + +Add to existing service's supervisord config: + +```ini +[program:meta-core] +command=/usr/local/bin/meta-core +priority=10 +autostart=true +autorestart=true +``` + +### Client Usage (TypeScript) + +```typescript +const response = await fetch('http://localhost:9000/meta/' + hashId); +const { metadata } = await response.json(); +``` + +### Client Usage (Python) + +```python +import requests +response = requests.get(f'http://localhost:9000/meta/{hash_id}') +metadata = response.json()['metadata'] +``` + +## Dependencies + +| Dependency | Version | Purpose | +|------------|---------|---------| +| `github.com/gorilla/mux` | v1.8.1 | HTTP router | +| `github.com/redis/go-redis/v9` | v9.4.0 | Redis client | +| Go stdlib | - | context, encoding/json, net, os, sync, syscall | + +### Runtime Requirements + +- **Go 1.21+** for building +- **Redis** binary available in PATH (leader spawns it) +- **Shared filesystem** accessible by all instances (for flock) + +## License + +Part of the MetaMesh project. diff --git a/cmd/meta-core/main.go b/cmd/meta-core/main.go new file mode 100644 index 0000000..c820ec6 --- /dev/null +++ b/cmd/meta-core/main.go @@ -0,0 +1,111 @@ +package main + +import ( + "log" + "os" + "os/signal" + "syscall" + + "github.com/metazla/meta-core/internal/api" + "github.com/metazla/meta-core/internal/config" + "github.com/metazla/meta-core/internal/discovery" + "github.com/metazla/meta-core/internal/leader" + "github.com/metazla/meta-core/internal/storage" +) + +// Version is set at build time +var Version = "1.0.0" + +func main() { + log.SetFlags(log.LstdFlags | log.Lshortfile) + log.Printf("[meta-core] Starting version %s", Version) + + // Load configuration + cfg := config.Load() + log.Printf("[meta-core] Service: %s, HTTP Port: %d", cfg.ServiceName, cfg.HTTPPort) + + // Create storage client + storageClient := storage.NewClient("") + + // Create leader election + election := leader.NewElection(cfg) + + // Wire up storage connector to election + election.SetStorageConnector(&storageConnectorAdapter{client: storageClient}) + + // Set up leader/follower callbacks + election.OnBecomeLeader(func() { + log.Println("[meta-core] Became LEADER - Redis is running") + }) + + election.OnBecomeFollower(func(info *leader.LeaderLockInfo) { + log.Printf("[meta-core] Became FOLLOWER - Leader at %s", info.API) + }) + + election.OnLeaderLost(func() { + log.Println("[meta-core] Lost leadership") + }) + + // Start leader election + if err := election.Start(); err != nil { + log.Fatalf("[meta-core] Failed to start leader election: %v", err) + } + + // Create and start service discovery + disc := discovery.NewService(cfg) + if err := disc.Start(); err != nil { + log.Fatalf("[meta-core] Failed to start service discovery: %v", err) + } + + // Create and start API server + apiServer := api.NewServer(cfg, election, disc, storageClient) + if err := apiServer.Start(); err != nil { + log.Fatalf("[meta-core] Failed to start API server: %v", err) + } + + log.Println("[meta-core] Ready and serving requests") + + // Wait for shutdown signal + waitForShutdown() + + log.Println("[meta-core] Shutting down...") + + // Graceful shutdown in reverse order + if err := apiServer.Stop(); err != nil { + log.Printf("[meta-core] Error stopping API server: %v", err) + } + + if err := disc.Stop(); err != nil { + log.Printf("[meta-core] Error stopping service discovery: %v", err) + } + + if err := election.Stop(); err != nil { + log.Printf("[meta-core] Error stopping leader election: %v", err) + } + + if err := storageClient.Close(); err != nil { + log.Printf("[meta-core] Error closing storage client: %v", err) + } + + log.Println("[meta-core] Shutdown complete") +} + +// waitForShutdown blocks until a shutdown signal is received +func waitForShutdown() { + sigChan := make(chan os.Signal, 1) + signal.Notify(sigChan, syscall.SIGINT, syscall.SIGTERM) + <-sigChan +} + +// storageConnectorAdapter adapts storage.Client to leader.StorageConnector +type storageConnectorAdapter struct { + client *storage.Client +} + +func (a *storageConnectorAdapter) Connect(url string) error { + return a.client.Connect(url) +} + +func (a *storageConnectorAdapter) Close() error { + return a.client.Close() +} diff --git a/go.mod b/go.mod new file mode 100644 index 0000000..539a3ee --- /dev/null +++ b/go.mod @@ -0,0 +1,13 @@ +module github.com/metazla/meta-core + +go 1.21 + +require ( + github.com/gorilla/mux v1.8.1 + github.com/redis/go-redis/v9 v9.4.0 +) + +require ( + github.com/cespare/xxhash/v2 v2.2.0 // indirect + github.com/dgryski/go-rendezvous v0.0.0-20200823014737-9f7001d12a5f // indirect +) diff --git a/go.sum b/go.sum new file mode 100644 index 0000000..d169c82 --- /dev/null +++ b/go.sum @@ -0,0 +1,8 @@ +github.com/cespare/xxhash/v2 v2.2.0 h1:DC2CZ1Ep5Y4k3ZQ899DldepgrayRUGE6BBZ/cd9Cj44= +github.com/cespare/xxhash/v2 v2.2.0/go.mod h1:VGX0DQ3Q6kWi7AoAeZDth3/j3BFtOZR5XLFGgcrjCOs= +github.com/dgryski/go-rendezvous v0.0.0-20200823014737-9f7001d12a5f h1:lO4WD4F/rVNCu3HqELle0jiPLLBs70cWOduZpkS1E78= +github.com/dgryski/go-rendezvous v0.0.0-20200823014737-9f7001d12a5f/go.mod h1:cuUVRXasLTGF7a8hSLbxyZXjz+1KgoB3wDUb6vlszIc= +github.com/gorilla/mux v1.8.1 h1:TuBL49tXwgrFYWhqrNgrUNEY92u81SPhu7sTdzQEiWY= +github.com/gorilla/mux v1.8.1/go.mod h1:AKf9I4AEqPTmMytcMc0KkNouC66V3BtZ4qD5fmWSiMQ= +github.com/redis/go-redis/v9 v9.4.0 h1:Yzoz33UZw9I/mFhx4MNrB6Fk+XHO1VukNcCa1+lwyKk= +github.com/redis/go-redis/v9 v9.4.0/go.mod h1:hdY0cQFCN4fnSYT6TkisLufl/4W5UIXyv0b/CLO2V2M= diff --git a/internal/api/handlers.go b/internal/api/handlers.go new file mode 100644 index 0000000..8e9886a --- /dev/null +++ b/internal/api/handlers.go @@ -0,0 +1,720 @@ +package api + +import ( + "crypto/sha256" + "encoding/base32" + "encoding/json" + "io" + "net/http" + "os" + "path/filepath" + "strings" + "time" + + "github.com/gorilla/mux" + "github.com/metazla/meta-core/internal/leader" +) + +// contentTypeByExt maps file extensions to MIME types +var contentTypeByExt = map[string]string{ + ".jpg": "image/jpeg", + ".jpeg": "image/jpeg", + ".png": "image/png", + ".webp": "image/webp", + ".gif": "image/gif", + ".mp4": "video/mp4", + ".mkv": "video/x-matroska", + ".avi": "video/x-msvideo", + ".webm": "video/webm", + ".mov": "video/quicktime", + ".ts": "video/mp2t", + ".m3u8": "application/vnd.apple.mpegurl", +} + +// HealthResponse is the response for /health +type HealthResponse struct { + Status string `json:"status"` + Role string `json:"role"` + Redis bool `json:"redis"` + Timestamp string `json:"timestamp"` + Leader *leader.LeaderLockInfo `json:"leader,omitempty"` +} + +// StatusResponse is the response for /status +type StatusResponse struct { + Status string `json:"status"` + Role string `json:"role"` + Redis bool `json:"redis"` + ServiceName string `json:"serviceName"` + Version string `json:"version"` + Uptime int64 `json:"uptimeSeconds"` + FileCount int `json:"fileCount"` + Leader *leader.LeaderLockInfo `json:"leader,omitempty"` +} + +// MetadataResponse is the response for /meta/{hash} +type MetadataResponse struct { + HashID string `json:"hashId"` + Metadata map[string]string `json:"metadata"` +} + +// DataPathResponse is the response for /data/{hash}/path +type DataPathResponse struct { + HashID string `json:"hashId"` + Path string `json:"path"` + Exists bool `json:"exists"` +} + +// ErrorResponse is the response for errors +type ErrorResponse struct { + Error string `json:"error"` + Message string `json:"message,omitempty"` +} + +var startTime = time.Now() + +// handleHealth handles GET /health +func (s *Server) handleHealth(w http.ResponseWriter, r *http.Request) { + response := HealthResponse{ + Status: "ok", + Role: string(s.election.Role()), + Redis: s.storage.Health(), + Timestamp: time.Now().UTC().Format(time.RFC3339), + Leader: s.election.LeaderInfo(), + } + + if !response.Redis { + response.Status = "degraded" + } + + writeJSON(w, http.StatusOK, response) +} + +// handleStatus handles GET /status +func (s *Server) handleStatus(w http.ResponseWriter, r *http.Request) { + fileCount := 0 + if s.storage.IsConnected() { + count, _ := s.storage.CountFiles() + fileCount = count + } + + response := StatusResponse{ + Status: "ok", + Role: string(s.election.Role()), + Redis: s.storage.Health(), + ServiceName: s.config.ServiceName, + Version: s.config.ServiceVersion, + Uptime: int64(time.Since(startTime).Seconds()), + FileCount: fileCount, + Leader: s.election.LeaderInfo(), + } + + if !response.Redis { + response.Status = "degraded" + } + + writeJSON(w, http.StatusOK, response) +} + +// handleLeader handles GET /leader +func (s *Server) handleLeader(w http.ResponseWriter, r *http.Request) { + info := s.election.LeaderInfo() + if info == nil { + writeError(w, http.StatusServiceUnavailable, "no leader available") + return + } + + writeJSON(w, http.StatusOK, info) +} + +// handleRole handles GET /role +func (s *Server) handleRole(w http.ResponseWriter, r *http.Request) { + writeJSON(w, http.StatusOK, map[string]string{ + "role": string(s.election.Role()), + }) +} + +// handleGetMeta handles GET /meta/{hash} +func (s *Server) handleGetMeta(w http.ResponseWriter, r *http.Request) { + vars := mux.Vars(r) + hashID := vars["hash"] + + if hashID == "" { + writeError(w, http.StatusBadRequest, "hash is required") + return + } + + if !s.storage.IsConnected() { + writeError(w, http.StatusServiceUnavailable, "storage not connected") + return + } + + metadata, err := s.storage.GetMetadataFlat(hashID) + if err != nil { + writeError(w, http.StatusInternalServerError, err.Error()) + return + } + + if metadata == nil { + writeError(w, http.StatusNotFound, "metadata not found") + return + } + + response := MetadataResponse{ + HashID: hashID, + Metadata: metadata, + } + + writeJSON(w, http.StatusOK, response) +} + +// handlePutMeta handles PUT /meta/{hash} +func (s *Server) handlePutMeta(w http.ResponseWriter, r *http.Request) { + vars := mux.Vars(r) + hashID := vars["hash"] + + if hashID == "" { + writeError(w, http.StatusBadRequest, "hash is required") + return + } + + if !s.storage.IsConnected() { + writeError(w, http.StatusServiceUnavailable, "storage not connected") + return + } + + var metadata map[string]string + if err := json.NewDecoder(r.Body).Decode(&metadata); err != nil { + writeError(w, http.StatusBadRequest, "invalid JSON body") + return + } + + if err := s.storage.SetMetadataFlat(hashID, metadata); err != nil { + writeError(w, http.StatusInternalServerError, err.Error()) + return + } + + writeJSON(w, http.StatusOK, map[string]interface{}{ + "success": true, + "hashId": hashID, + }) +} + +// handleDeleteMeta handles DELETE /meta/{hash} +func (s *Server) handleDeleteMeta(w http.ResponseWriter, r *http.Request) { + vars := mux.Vars(r) + hashID := vars["hash"] + + if hashID == "" { + writeError(w, http.StatusBadRequest, "hash is required") + return + } + + if !s.storage.IsConnected() { + writeError(w, http.StatusServiceUnavailable, "storage not connected") + return + } + + deleted, err := s.storage.DeleteMetadata(hashID) + if err != nil { + writeError(w, http.StatusInternalServerError, err.Error()) + return + } + + writeJSON(w, http.StatusOK, map[string]interface{}{ + "success": true, + "hashId": hashID, + "deleted": deleted, + }) +} + +// handleListMeta handles GET /meta +func (s *Server) handleListMeta(w http.ResponseWriter, r *http.Request) { + if !s.storage.IsConnected() { + writeError(w, http.StatusServiceUnavailable, "storage not connected") + return + } + + hashIDs, err := s.storage.GetAllHashIDs() + if err != nil { + writeError(w, http.StatusInternalServerError, err.Error()) + return + } + + writeJSON(w, http.StatusOK, map[string]interface{}{ + "hashIds": hashIDs, + "count": len(hashIDs), + }) +} + +// handleGetDataPath handles GET /data/{hash}/path +func (s *Server) handleGetDataPath(w http.ResponseWriter, r *http.Request) { + vars := mux.Vars(r) + hashID := vars["hash"] + + if hashID == "" { + writeError(w, http.StatusBadRequest, "hash is required") + return + } + + if !s.storage.IsConnected() { + writeError(w, http.StatusServiceUnavailable, "storage not connected") + return + } + + // Get the file path from metadata + filePath, err := s.storage.GetProperty(hashID, "filePath") + if err != nil { + writeError(w, http.StatusInternalServerError, err.Error()) + return + } + + if filePath == "" { + writeError(w, http.StatusNotFound, "file path not found") + return + } + + // Check if file exists + fullPath := s.config.FilesPath + "/" + filePath + _, statErr := os.Stat(fullPath) + exists := statErr == nil + + response := DataPathResponse{ + HashID: hashID, + Path: fullPath, + Exists: exists, + } + + writeJSON(w, http.StatusOK, response) +} + +// handleHeadData handles HEAD /data/{hash} +func (s *Server) handleHeadData(w http.ResponseWriter, r *http.Request) { + vars := mux.Vars(r) + hashID := vars["hash"] + + if hashID == "" { + w.WriteHeader(http.StatusBadRequest) + return + } + + if !s.storage.IsConnected() { + w.WriteHeader(http.StatusServiceUnavailable) + return + } + + // Get the file path from metadata + filePath, err := s.storage.GetProperty(hashID, "filePath") + if err != nil || filePath == "" { + w.WriteHeader(http.StatusNotFound) + return + } + + // Check if file exists + fullPath := s.config.FilesPath + "/" + filePath + if _, err := os.Stat(fullPath); err != nil { + w.WriteHeader(http.StatusNotFound) + return + } + + w.WriteHeader(http.StatusOK) +} + +// handleGetFileByCID handles GET /file/{cid} +// Serves a file by looking up its CID in poster/backdrop metadata +func (s *Server) handleGetFileByCID(w http.ResponseWriter, r *http.Request) { + vars := mux.Vars(r) + cid := vars["cid"] + + if cid == "" { + writeError(w, http.StatusBadRequest, "cid is required") + return + } + + if !s.storage.IsConnected() { + writeError(w, http.StatusServiceUnavailable, "storage not connected") + return + } + + // Look up the file path by CID + relPath, err := s.storage.LookupPathByCID(cid) + if err != nil { + writeError(w, http.StatusInternalServerError, err.Error()) + return + } + + if relPath == "" { + writeError(w, http.StatusNotFound, "file not found for CID") + return + } + + // Construct full path + fullPath := filepath.Join(s.config.FilesPath, relPath) + + // Check if file exists + fileInfo, err := os.Stat(fullPath) + if err != nil { + writeError(w, http.StatusNotFound, "file does not exist on disk") + return + } + + // Determine content type from extension + ext := strings.ToLower(filepath.Ext(fullPath)) + contentType, ok := contentTypeByExt[ext] + if !ok { + contentType = "application/octet-stream" + } + + // Open the file + file, err := os.Open(fullPath) + if err != nil { + writeError(w, http.StatusInternalServerError, "failed to open file") + return + } + defer file.Close() + + // Set content type header + w.Header().Set("Content-Type", contentType) + + // Serve the file (supports range requests) + http.ServeContent(w, r, fileInfo.Name(), fileInfo.ModTime(), file) +} + +// handleListServices handles GET /services +func (s *Server) handleListServices(w http.ResponseWriter, r *http.Request) { + services, err := s.discovery.DiscoverAll() + if err != nil { + writeError(w, http.StatusInternalServerError, err.Error()) + return + } + + writeJSON(w, http.StatusOK, map[string]interface{}{ + "services": services, + "count": len(services), + }) +} + +// handleGetService handles GET /services/{name} +func (s *Server) handleGetService(w http.ResponseWriter, r *http.Request) { + vars := mux.Vars(r) + name := vars["name"] + + if name == "" { + writeError(w, http.StatusBadRequest, "service name is required") + return + } + + service, err := s.discovery.Discover(name) + if err != nil { + writeError(w, http.StatusInternalServerError, err.Error()) + return + } + + if service == nil { + writeError(w, http.StatusNotFound, "service not found") + return + } + + writeJSON(w, http.StatusOK, service) +} + +// writeJSON writes a JSON response +func writeJSON(w http.ResponseWriter, status int, data interface{}) { + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(status) + json.NewEncoder(w).Encode(data) +} + +// writeError writes an error response +func writeError(w http.ResponseWriter, status int, message string) { + writeJSON(w, status, ErrorResponse{ + Error: http.StatusText(status), + Message: message, + }) +} + +// handlePatchMeta handles PATCH /meta/{hash} +// Merges the provided metadata into existing (does not delete missing keys) +func (s *Server) handlePatchMeta(w http.ResponseWriter, r *http.Request) { + vars := mux.Vars(r) + hashID := vars["hash"] + + if hashID == "" { + writeError(w, http.StatusBadRequest, "hash is required") + return + } + + if !s.storage.IsConnected() { + writeError(w, http.StatusServiceUnavailable, "storage not connected") + return + } + + var metadata map[string]string + if err := json.NewDecoder(r.Body).Decode(&metadata); err != nil { + writeError(w, http.StatusBadRequest, "invalid JSON body") + return + } + + updated, err := s.storage.MergeMetadataFlat(hashID, metadata) + if err != nil { + writeError(w, http.StatusInternalServerError, err.Error()) + return + } + + writeJSON(w, http.StatusOK, map[string]interface{}{ + "success": true, + "hashId": hashID, + "updated": updated, + }) +} + +// handleGetProperty handles GET /meta/{hash}/{key} +// Gets a single property value +func (s *Server) handleGetProperty(w http.ResponseWriter, r *http.Request) { + vars := mux.Vars(r) + hashID := vars["hash"] + key := vars["key"] + + if hashID == "" { + writeError(w, http.StatusBadRequest, "hash is required") + return + } + + if key == "" { + writeError(w, http.StatusBadRequest, "key is required") + return + } + + if !s.storage.IsConnected() { + writeError(w, http.StatusServiceUnavailable, "storage not connected") + return + } + + value, err := s.storage.GetProperty(hashID, key) + if err != nil { + writeError(w, http.StatusInternalServerError, err.Error()) + return + } + + if value == "" { + writeError(w, http.StatusNotFound, "property not found") + return + } + + // Return plain text for single property + w.Header().Set("Content-Type", "text/plain") + w.WriteHeader(http.StatusOK) + w.Write([]byte(value)) +} + +// handlePutProperty handles PUT /meta/{hash}/{key} +// Sets a single property value +func (s *Server) handlePutProperty(w http.ResponseWriter, r *http.Request) { + vars := mux.Vars(r) + hashID := vars["hash"] + key := vars["key"] + + if hashID == "" { + writeError(w, http.StatusBadRequest, "hash is required") + return + } + + if key == "" { + writeError(w, http.StatusBadRequest, "key is required") + return + } + + if !s.storage.IsConnected() { + writeError(w, http.StatusServiceUnavailable, "storage not connected") + return + } + + // Read body as plain text + buf := make([]byte, 1024*1024) // 1MB max + n, err := r.Body.Read(buf) + if err != nil && err.Error() != "EOF" { + writeError(w, http.StatusBadRequest, "failed to read body") + return + } + value := string(buf[:n]) + + if err := s.storage.SetProperty(hashID, key, value); err != nil { + writeError(w, http.StatusInternalServerError, err.Error()) + return + } + + writeJSON(w, http.StatusOK, map[string]interface{}{ + "success": true, + "hashId": hashID, + "property": key, + }) +} + +// handleDeleteProperty handles DELETE /meta/{hash}/{key} +// Deletes a single property +func (s *Server) handleDeleteProperty(w http.ResponseWriter, r *http.Request) { + vars := mux.Vars(r) + hashID := vars["hash"] + key := vars["key"] + + if hashID == "" { + writeError(w, http.StatusBadRequest, "hash is required") + return + } + + if key == "" { + writeError(w, http.StatusBadRequest, "key is required") + return + } + + if !s.storage.IsConnected() { + writeError(w, http.StatusServiceUnavailable, "storage not connected") + return + } + + if err := s.storage.DeleteProperty(hashID, key); err != nil { + writeError(w, http.StatusInternalServerError, err.Error()) + return + } + + writeJSON(w, http.StatusOK, map[string]interface{}{ + "success": true, + "hashId": hashID, + "property": key, + }) +} + +// handleAddToSet handles POST /meta/{hash}/_add/{key} +// Adds a value to a set-type field +func (s *Server) handleAddToSet(w http.ResponseWriter, r *http.Request) { + vars := mux.Vars(r) + hashID := vars["hash"] + key := vars["key"] + + if hashID == "" { + writeError(w, http.StatusBadRequest, "hash is required") + return + } + + if key == "" { + writeError(w, http.StatusBadRequest, "key is required") + return + } + + if !s.storage.IsConnected() { + writeError(w, http.StatusServiceUnavailable, "storage not connected") + return + } + + // Read body as plain text + buf := make([]byte, 1024*1024) // 1MB max + n, err := r.Body.Read(buf) + if err != nil && err.Error() != "EOF" { + writeError(w, http.StatusBadRequest, "failed to read body") + return + } + value := string(buf[:n]) + + if value == "" { + writeError(w, http.StatusBadRequest, "value is required") + return + } + + added, err := s.storage.AddToSet(hashID, key, value) + if err != nil { + writeError(w, http.StatusInternalServerError, err.Error()) + return + } + + writeJSON(w, http.StatusOK, map[string]interface{}{ + "success": true, + "hashId": hashID, + "property": key, + "added": added, + }) +} + +// CIDRequest is the request body for POST /file/cid +type CIDRequest struct { + Path string `json:"path"` +} + +// CIDResponse is the response for POST /file/cid +type CIDResponse struct { + CID string `json:"cid"` + Path string `json:"path"` + Size int64 `json:"size"` +} + +// handleComputeFileCID handles POST /file/cid +// Computes the IPFS-compatible CIDv1 (sha256) for a file +func (s *Server) handleComputeFileCID(w http.ResponseWriter, r *http.Request) { + var req CIDRequest + if err := json.NewDecoder(r.Body).Decode(&req); err != nil { + writeError(w, http.StatusBadRequest, "invalid JSON body") + return + } + + if req.Path == "" { + writeError(w, http.StatusBadRequest, "path is required") + return + } + + // Construct full path - path should be relative to FILES_PATH + fullPath := filepath.Join(s.config.FilesPath, req.Path) + + // Security check: ensure path is within FILES_PATH + absFilesPath, _ := filepath.Abs(s.config.FilesPath) + absFullPath, _ := filepath.Abs(fullPath) + if !strings.HasPrefix(absFullPath, absFilesPath) { + writeError(w, http.StatusBadRequest, "path must be within files directory") + return + } + + // Check if file exists + fileInfo, err := os.Stat(fullPath) + if err != nil { + writeError(w, http.StatusNotFound, "file not found") + return + } + + if fileInfo.IsDir() { + writeError(w, http.StatusBadRequest, "path is a directory, not a file") + return + } + + // Open and hash the file + file, err := os.Open(fullPath) + if err != nil { + writeError(w, http.StatusInternalServerError, "failed to open file") + return + } + defer file.Close() + + // Compute SHA-256 hash + hasher := sha256.New() + if _, err := io.Copy(hasher, file); err != nil { + writeError(w, http.StatusInternalServerError, "failed to read file") + return + } + hashBytes := hasher.Sum(nil) + + // Build CIDv1 with raw codec (0x55) and sha256 multihash + // CIDv1 format: version (0x01) + codec (0x55) + multihash + // Multihash format: hash-code (0x12) + length (0x20) + hash + cidBytes := make([]byte, 0, 2+2+32) + cidBytes = append(cidBytes, 0x01) // CIDv1 + cidBytes = append(cidBytes, 0x55) // raw codec + cidBytes = append(cidBytes, 0x12) // sha256 code + cidBytes = append(cidBytes, 0x20) // 32 bytes + cidBytes = append(cidBytes, hashBytes...) + + // Encode as base32lower with 'b' prefix (multibase) + cid := "b" + strings.ToLower(base32.StdEncoding.WithPadding(base32.NoPadding).EncodeToString(cidBytes)) + + writeJSON(w, http.StatusOK, CIDResponse{ + CID: cid, + Path: req.Path, + Size: fileInfo.Size(), + }) +} diff --git a/internal/api/server.go b/internal/api/server.go new file mode 100644 index 0000000..c86f17b --- /dev/null +++ b/internal/api/server.go @@ -0,0 +1,146 @@ +package api + +import ( + "context" + "fmt" + "log" + "net/http" + "time" + + "github.com/gorilla/mux" + "github.com/metazla/meta-core/internal/config" + "github.com/metazla/meta-core/internal/discovery" + "github.com/metazla/meta-core/internal/leader" + "github.com/metazla/meta-core/internal/storage" +) + +// Server is the HTTP API server for meta-core +type Server struct { + config *config.Config + election *leader.Election + discovery *discovery.Service + storage *storage.Client + router *mux.Router + server *http.Server +} + +// NewServer creates a new API server +func NewServer( + cfg *config.Config, + election *leader.Election, + disc *discovery.Service, + stor *storage.Client, +) *Server { + s := &Server{ + config: cfg, + election: election, + discovery: disc, + storage: stor, + router: mux.NewRouter(), + } + + s.setupRoutes() + return s +} + +// setupRoutes configures all API routes +func (s *Server) setupRoutes() { + // Health and status + s.router.HandleFunc("/health", s.handleHealth).Methods("GET") + s.router.HandleFunc("/status", s.handleStatus).Methods("GET") + s.router.HandleFunc("/leader", s.handleLeader).Methods("GET") + s.router.HandleFunc("/role", s.handleRole).Methods("GET") + + // Metadata operations - base endpoints + s.router.HandleFunc("/meta/{hash}", s.handleGetMeta).Methods("GET") + s.router.HandleFunc("/meta/{hash}", s.handlePutMeta).Methods("PUT") + s.router.HandleFunc("/meta/{hash}", s.handlePatchMeta).Methods("PATCH") + s.router.HandleFunc("/meta/{hash}", s.handleDeleteMeta).Methods("DELETE") + s.router.HandleFunc("/meta", s.handleListMeta).Methods("GET") + + // Metadata operations - set operations (must be before property routes) + s.router.HandleFunc("/meta/{hash}/_add/{key:.*}", s.handleAddToSet).Methods("POST") + + // Metadata operations - property-level (key can contain slashes) + s.router.HandleFunc("/meta/{hash}/{key:.*}", s.handleGetProperty).Methods("GET") + s.router.HandleFunc("/meta/{hash}/{key:.*}", s.handlePutProperty).Methods("PUT") + s.router.HandleFunc("/meta/{hash}/{key:.*}", s.handleDeleteProperty).Methods("DELETE") + + // Data operations + s.router.HandleFunc("/data/{hash}/path", s.handleGetDataPath).Methods("GET") + s.router.HandleFunc("/data/{hash}", s.handleHeadData).Methods("HEAD") + + // File operations (by CID) + s.router.HandleFunc("/file/{cid}", s.handleGetFileByCID).Methods("GET") + s.router.HandleFunc("/file/cid", s.handleComputeFileCID).Methods("POST") + + // Service discovery + s.router.HandleFunc("/services", s.handleListServices).Methods("GET") + s.router.HandleFunc("/services/{name}", s.handleGetService).Methods("GET") + + // Add middleware + s.router.Use(loggingMiddleware) + s.router.Use(corsMiddleware) +} + +// Start starts the HTTP server +func (s *Server) Start() error { + addr := fmt.Sprintf("%s:%d", s.config.HTTPHost, s.config.HTTPPort) + + s.server = &http.Server{ + Addr: addr, + Handler: s.router, + ReadTimeout: 30 * time.Second, + WriteTimeout: 30 * time.Second, + IdleTimeout: 60 * time.Second, + } + + log.Printf("[API] Starting HTTP server on %s", addr) + + go func() { + if err := s.server.ListenAndServe(); err != nil && err != http.ErrServerClosed { + log.Printf("[API] HTTP server error: %v", err) + } + }() + + return nil +} + +// Stop gracefully stops the HTTP server +func (s *Server) Stop() error { + if s.server == nil { + return nil + } + + log.Println("[API] Stopping HTTP server...") + + ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) + defer cancel() + + return s.server.Shutdown(ctx) +} + +// loggingMiddleware logs all requests +func loggingMiddleware(next http.Handler) http.Handler { + return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + start := time.Now() + next.ServeHTTP(w, r) + log.Printf("[API] %s %s %s", r.Method, r.URL.Path, time.Since(start)) + }) +} + +// corsMiddleware adds CORS headers +func corsMiddleware(next http.Handler) http.Handler { + return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.Header().Set("Access-Control-Allow-Origin", "*") + w.Header().Set("Access-Control-Allow-Methods", "GET, POST, PUT, DELETE, OPTIONS") + w.Header().Set("Access-Control-Allow-Headers", "Content-Type") + + if r.Method == "OPTIONS" { + w.WriteHeader(http.StatusOK) + return + } + + next.ServeHTTP(w, r) + }) +} diff --git a/internal/config/config.go b/internal/config/config.go new file mode 100644 index 0000000..3da4548 --- /dev/null +++ b/internal/config/config.go @@ -0,0 +1,85 @@ +package config + +import ( + "os" + "strconv" +) + +// Config holds all configuration for meta-core +type Config struct { + // Core paths + MetaCorePath string // Path to meta-core volume (default: /meta-core) + FilesPath string // Path to files volume (default: /files) + + // Service identification + ServiceName string // Service name (e.g., "meta-sort", "meta-fuse") + ServiceVersion string // Service version (default: "1.0.0") + APIPort int // Service HTTP port (for leader info) + BaseURL string // Base URL for stable service discovery + + // Redis configuration + RedisPort int // Redis port (default: 6379) + + // HTTP API configuration + HTTPPort int // HTTP API port (default: 9000) + HTTPHost string // HTTP API host (default: "127.0.0.1") + + // Timing configuration + HealthCheckIntervalMS int // Health check interval in ms (default: 5000) + HeartbeatIntervalMS int // Service heartbeat interval in ms (default: 30000) + StaleThresholdMS int // Stale service threshold in ms (default: 60000) +} + +// Load creates a Config from environment variables +func Load() *Config { + return &Config{ + MetaCorePath: getEnv("META_CORE_PATH", "/meta-core"), + FilesPath: getEnv("FILES_PATH", "/files"), + ServiceName: getEnv("SERVICE_NAME", "meta-core"), + ServiceVersion: getEnv("SERVICE_VERSION", "1.0.0"), + APIPort: getEnvInt("API_PORT", 8180), + BaseURL: getEnv("BASE_URL", ""), + RedisPort: getEnvInt("REDIS_PORT", 6379), + HTTPPort: getEnvInt("META_CORE_HTTP_PORT", 9000), + HTTPHost: getEnv("META_CORE_HTTP_HOST", "127.0.0.1"), + HealthCheckIntervalMS: getEnvInt("HEALTH_CHECK_INTERVAL_MS", 5000), + HeartbeatIntervalMS: getEnvInt("HEARTBEAT_INTERVAL_MS", 30000), + StaleThresholdMS: getEnvInt("STALE_THRESHOLD_MS", 60000), + } +} + +// LockFilePath returns the path to the leader lock file +func (c *Config) LockFilePath() string { + return c.MetaCorePath + "/locks/kv-leader.lock" +} + +// InfoFilePath returns the path to the leader info file +func (c *Config) InfoFilePath() string { + return c.MetaCorePath + "/locks/kv-leader.info" +} + +// RedisDataDir returns the path to Redis data directory +func (c *Config) RedisDataDir() string { + return c.MetaCorePath + "/db/redis" +} + +// ServicesDir returns the path to services directory +func (c *Config) ServicesDir() string { + return c.MetaCorePath + "/services" +} + +func getEnv(key, defaultValue string) string { + if value := os.Getenv(key); value != "" { + return value + } + return defaultValue +} + +func getEnvInt(key string, defaultValue int) int { + if value := os.Getenv(key); value != "" { + if intVal, err := strconv.Atoi(value); err == nil { + return intVal + } + } + return defaultValue +} diff --git a/internal/discovery/service.go b/internal/discovery/service.go new file mode 100644 index 0000000..30b2699 --- /dev/null +++ b/internal/discovery/service.go @@ -0,0 +1,258 @@ +package discovery + +import ( + "encoding/json" + "fmt" + "log" + "net" + "os" + "path/filepath" + "sync" + "time" + + "github.com/metazla/meta-core/internal/config" +) + +// ServiceInfo matches the TypeScript ServiceInfo interface +type ServiceInfo struct { + Name string `json:"name"` + Version string `json:"version"` + API string `json:"api"` + Status string `json:"status"` + PID int `json:"pid"` + Hostname string `json:"hostname"` + StartedAt string `json:"startedAt"` + LastHeartbeat string `json:"lastHeartbeat"` + Capabilities []string `json:"capabilities"` + Endpoints map[string]string `json:"endpoints"` +} + +// Service handles service registration and discovery +type Service struct { + config *config.Config + servicesDir string + serviceFile string + info *ServiceInfo + mu sync.RWMutex + + stopChan chan struct{} + wg sync.WaitGroup +} + +// NewService creates a new service discovery instance +func NewService(cfg *config.Config) *Service { + return &Service{ + config: cfg, + servicesDir: cfg.ServicesDir(), + serviceFile: filepath.Join(cfg.ServicesDir(), cfg.ServiceName+".json"), + stopChan: make(chan struct{}), + } +} + +// Start begins service registration and heartbeat +func (s *Service) Start() error { + log.Printf("[Discovery] Starting service discovery for %s", s.config.ServiceName) + + // Ensure services directory exists + if err := os.MkdirAll(s.servicesDir, 0755); err != nil { + return fmt.Errorf("failed to create services directory: %w", err) + } + + // Build and register service info + s.info = s.buildServiceInfo() + if err := s.register(); err != nil { + return fmt.Errorf("failed to register service: %w", err) + } + + // Start heartbeat loop + s.wg.Add(1) + go s.heartbeatLoop() + + return nil +} + +// Stop stops service discovery and unregisters +func (s *Service) Stop() error { + log.Println("[Discovery] Stopping service discovery...") + close(s.stopChan) + s.wg.Wait() + + // Unregister by removing service file + if err := os.Remove(s.serviceFile); err != nil && !os.IsNotExist(err) { + log.Printf("[Discovery] Failed to remove service file: %v", err) + } + + return nil +} + +// buildServiceInfo creates the service info for this instance +func (s *Service) buildServiceInfo() *ServiceInfo { + hostname, _ := os.Hostname() + ip := getLocalIP() + + return &ServiceInfo{ + Name: s.config.ServiceName, + Version: s.config.ServiceVersion, + API: fmt.Sprintf("http://%s:%d", ip, s.config.APIPort), + Status: "running", + PID: os.Getpid(), + Hostname: hostname, + StartedAt: time.Now().UTC().Format(time.RFC3339), + LastHeartbeat: time.Now().UTC().Format(time.RFC3339), + Capabilities: []string{"meta-core"}, + Endpoints: map[string]string{ + "health": fmt.Sprintf("http://%s:%d/health", ip, s.config.HTTPPort), + "meta": fmt.Sprintf("http://%s:%d/meta", ip, s.config.HTTPPort), + }, + } +} + +// register writes service info to file +func (s *Service) register() error { + s.mu.Lock() + defer s.mu.Unlock() + + return s.writeServiceInfo(s.info) +} + +// writeServiceInfo atomically writes service info to file +func (s *Service) writeServiceInfo(info *ServiceInfo) error { + tempPath := s.serviceFile + ".tmp" + + data, err := json.MarshalIndent(info, "", " ") + if err != nil { + return err + } + + if err := os.WriteFile(tempPath, data, 0644); err != nil { + return err + } + + return os.Rename(tempPath, s.serviceFile) +} + +// heartbeatLoop periodically updates the last heartbeat +func (s *Service) heartbeatLoop() { + defer s.wg.Done() + + ticker := time.NewTicker(time.Duration(s.config.HeartbeatIntervalMS) * time.Millisecond) + defer ticker.Stop() + + for { + select { + case <-s.stopChan: + return + case <-ticker.C: + s.heartbeat() + } + } +} + +// heartbeat updates the last heartbeat timestamp +func (s *Service) heartbeat() { + s.mu.Lock() + defer s.mu.Unlock() + + if s.info == nil { + return + } + + s.info.LastHeartbeat = time.Now().UTC().Format(time.RFC3339) + if err := s.writeServiceInfo(s.info); err != nil { + log.Printf("[Discovery] Failed to update heartbeat: %v", err) + } +} + +// UpdateStatus updates the service status +func (s *Service) UpdateStatus(status string) error { + s.mu.Lock() + defer s.mu.Unlock() + + if s.info == nil { + return fmt.Errorf("service not registered") + } + + s.info.Status = status + s.info.LastHeartbeat = time.Now().UTC().Format(time.RFC3339) + return s.writeServiceInfo(s.info) +} + +// Discover finds a service by name +func (s *Service) Discover(name string) (*ServiceInfo, error) { + filePath := filepath.Join(s.servicesDir, name+".json") + + data, err := os.ReadFile(filePath) + if err != nil { + if os.IsNotExist(err) { + return nil, nil + } + return nil, err + } + + var info ServiceInfo + if err := json.Unmarshal(data, &info); err != nil { + return nil, err + } + + // Check if service is stale + lastHeartbeat, err := time.Parse(time.RFC3339, info.LastHeartbeat) + if err == nil { + staleThreshold := time.Duration(s.config.StaleThresholdMS) * time.Millisecond + if time.Since(lastHeartbeat) > staleThreshold { + info.Status = "stale" + } + } + + return &info, nil +} + +// DiscoverAll finds all registered services +func (s *Service) DiscoverAll() ([]*ServiceInfo, error) { + entries, err := os.ReadDir(s.servicesDir) + if err != nil { + if os.IsNotExist(err) { + return []*ServiceInfo{}, nil + } + return nil, err + } + + var services []*ServiceInfo + for _, entry := range entries { + if entry.IsDir() || filepath.Ext(entry.Name()) != ".json" { + continue + } + + name := entry.Name()[:len(entry.Name())-5] // Remove .json extension + info, err := s.Discover(name) + if err != nil { + log.Printf("[Discovery] Failed to read service %s: %v", name, err) + continue + } + + if info != nil { + services = append(services, info) + } + } + + return services, nil +} + +// getLocalIP returns the local IP address +func getLocalIP() string { + addrs, err := net.InterfaceAddrs() + if err != nil { + hostname, _ := os.Hostname() + return hostname + } + + for _, addr := range addrs { + if ipnet, ok := addr.(*net.IPNet); ok && !ipnet.IP.IsLoopback() { + if ipnet.IP.To4() != nil { + return ipnet.IP.String() + } + } + } + + hostname, _ := os.Hostname() + return hostname +} diff --git a/internal/leader/election.go b/internal/leader/election.go new file mode 100644 index 0000000..0d5f878 --- /dev/null +++ b/internal/leader/election.go @@ -0,0 +1,430 @@ +package leader + +import ( + "encoding/json" + "fmt" + "log" + "net" + "os" + "path/filepath" + "sync" + "syscall" + "time" + + "github.com/metazla/meta-core/internal/config" +) + +// LeaderLockInfo matches the TypeScript LeaderLockInfo interface +type LeaderLockInfo struct { + Host string `json:"host"` + API string `json:"api"` + HTTP string `json:"http"` + BaseURL string `json:"baseUrl,omitempty"` + Timestamp int64 `json:"timestamp"` + PID int `json:"pid"` +} + +// Role represents the current role of this instance +type Role string + +const ( + RoleUnknown Role = "unknown" + RoleLeader Role = "leader" + RoleFollower Role = "follower" +) + +// Election handles leader election using flock on shared filesystem +type Election struct { + config *config.Config + + lockFile *os.File + role Role + leaderInfo *LeaderLockInfo + mu sync.RWMutex + + redisManager *RedisManager + storage StorageConnector + + // Callbacks + onBecomeLeader func() + onBecomeFollower func(info *LeaderLockInfo) + onLeaderLost func() + + // Lifecycle + stopChan chan struct{} + isShuttingDown bool + wg sync.WaitGroup +} + +// StorageConnector interface for connecting to Redis +type StorageConnector interface { + Connect(url string) error + Close() error +} + +// NewElection creates a new leader election instance +func NewElection(cfg *config.Config) *Election { + return &Election{ + config: cfg, + role: RoleUnknown, + redisManager: NewRedisManager(cfg), + stopChan: make(chan struct{}), + } +} + +// SetStorageConnector sets the storage connector for Redis connections +func (e *Election) SetStorageConnector(s StorageConnector) { + e.storage = s +} + +// OnBecomeLeader sets callback for when this instance becomes leader +func (e *Election) OnBecomeLeader(fn func()) { + e.onBecomeLeader = fn +} + +// OnBecomeFollower sets callback for when this instance becomes follower +func (e *Election) OnBecomeFollower(fn func(info *LeaderLockInfo)) { + e.onBecomeFollower = fn +} + +// OnLeaderLost sets callback for when leadership is lost +func (e *Election) OnLeaderLost(fn func()) { + e.onLeaderLost = fn +} + +// Start begins the leader election process +func (e *Election) Start() error { + log.Println("[Election] Starting leader election...") + + // Ensure lock directory exists + lockDir := filepath.Dir(e.config.LockFilePath()) + if err := os.MkdirAll(lockDir, 0755); err != nil { + return fmt.Errorf("failed to create lock directory: %w", err) + } + + // Try to acquire the lock + acquired, err := e.tryAcquireLock() + if err != nil { + return fmt.Errorf("failed to acquire lock: %w", err) + } + + if acquired { + if err := e.transitionToLeader(); err != nil { + return err + } + } else { + if err := e.transitionToFollower(); err != nil { + return err + } + } + + // Start health check loop + e.wg.Add(1) + go e.healthCheckLoop() + + return nil +} + +// Stop gracefully shuts down the election +func (e *Election) Stop() error { + log.Println("[Election] Stopping leader election...") + e.isShuttingDown = true + close(e.stopChan) + + e.wg.Wait() + + // Stop Redis if we were leader + if e.Role() == RoleLeader { + if err := e.redisManager.Stop(); err != nil { + log.Printf("[Election] Error stopping Redis: %v", err) + } + } + + // Release the lock + e.releaseLock() + + return nil +} + +// Role returns the current role (thread-safe) +func (e *Election) Role() Role { + e.mu.RLock() + defer e.mu.RUnlock() + return e.role +} + +// LeaderInfo returns the current leader info (thread-safe) +func (e *Election) LeaderInfo() *LeaderLockInfo { + e.mu.RLock() + defer e.mu.RUnlock() + if e.leaderInfo == nil { + return nil + } + // Return a copy + info := *e.leaderInfo + return &info +} + +// IsLeader returns true if this instance is the leader +func (e *Election) IsLeader() bool { + return e.Role() == RoleLeader +} + +// tryAcquireLock attempts to acquire exclusive flock on the lock file +func (e *Election) tryAcquireLock() (bool, error) { + lockPath := e.config.LockFilePath() + + // Open or create the lock file + f, err := os.OpenFile(lockPath, os.O_CREATE|os.O_RDWR, 0666) + if err != nil { + return false, fmt.Errorf("failed to open lock file: %w", err) + } + + // Try non-blocking exclusive flock + err = syscall.Flock(int(f.Fd()), syscall.LOCK_EX|syscall.LOCK_NB) + if err != nil { + f.Close() + if err == syscall.EWOULDBLOCK { + log.Println("[Election] Lock is held by another process") + return false, nil + } + return false, fmt.Errorf("flock failed: %w", err) + } + + e.lockFile = f + log.Printf("[Election] Acquired flock on %s", lockPath) + return true, nil +} + +// releaseLock releases the flock +func (e *Election) releaseLock() { + if e.lockFile != nil { + syscall.Flock(int(e.lockFile.Fd()), syscall.LOCK_UN) + e.lockFile.Close() + e.lockFile = nil + log.Println("[Election] Released flock") + } +} + +// transitionToLeader handles becoming the leader +func (e *Election) transitionToLeader() error { + log.Println("[Election] Transitioning to LEADER role") + + e.mu.Lock() + e.role = RoleLeader + e.mu.Unlock() + + // Start Redis + if err := e.redisManager.Start(); err != nil { + return fmt.Errorf("failed to start Redis: %w", err) + } + + // Wait for Redis to be ready + if err := e.redisManager.WaitForReady(30 * time.Second); err != nil { + return fmt.Errorf("Redis not ready: %w", err) + } + + // Build and write leader info + info := e.buildLeaderInfo() + e.mu.Lock() + e.leaderInfo = info + e.mu.Unlock() + + if err := e.writeLeaderInfo(info); err != nil { + return fmt.Errorf("failed to write leader info: %w", err) + } + + // Connect storage to local Redis + if e.storage != nil { + if err := e.storage.Connect(info.API); err != nil { + log.Printf("[Election] Warning: failed to connect storage: %v", err) + } + } + + // Notify callback + if e.onBecomeLeader != nil { + e.onBecomeLeader() + } + + log.Println("[Election] Now acting as LEADER") + return nil +} + +// transitionToFollower handles becoming a follower +func (e *Election) transitionToFollower() error { + log.Println("[Election] Transitioning to FOLLOWER role") + + e.mu.Lock() + e.role = RoleFollower + e.mu.Unlock() + + // Read leader info + info, err := e.readLeaderInfo() + if err != nil { + return fmt.Errorf("failed to read leader info: %w", err) + } + + e.mu.Lock() + e.leaderInfo = info + e.mu.Unlock() + + if info != nil { + log.Printf("[Election] Leader is at %s", info.API) + + // Connect storage to leader's Redis + if e.storage != nil { + if err := e.storage.Connect(info.API); err != nil { + log.Printf("[Election] Warning: failed to connect to leader: %v", err) + } + } + + // Notify callback + if e.onBecomeFollower != nil { + e.onBecomeFollower(info) + } + } + + log.Println("[Election] Now acting as FOLLOWER") + return nil +} + +// buildLeaderInfo creates leader info for this instance +func (e *Election) buildLeaderInfo() *LeaderLockInfo { + ip := getLocalIP() + hostname, _ := os.Hostname() + + return &LeaderLockInfo{ + Host: hostname, + API: fmt.Sprintf("redis://%s:%d", ip, e.config.RedisPort), + HTTP: fmt.Sprintf("http://%s:%d", ip, e.config.APIPort), + BaseURL: e.config.BaseURL, + Timestamp: time.Now().UnixMilli(), + PID: os.Getpid(), + } +} + +// writeLeaderInfo atomically writes leader info to file +func (e *Election) writeLeaderInfo(info *LeaderLockInfo) error { + infoPath := e.config.InfoFilePath() + tempPath := infoPath + ".tmp" + + data, err := json.MarshalIndent(info, "", " ") + if err != nil { + return err + } + + if err := os.WriteFile(tempPath, data, 0644); err != nil { + return err + } + + if err := os.Rename(tempPath, infoPath); err != nil { + return err + } + + log.Printf("[Election] Wrote leader info to %s", infoPath) + return nil +} + +// readLeaderInfo reads leader info from file +func (e *Election) readLeaderInfo() (*LeaderLockInfo, error) { + infoPath := e.config.InfoFilePath() + + data, err := os.ReadFile(infoPath) + if err != nil { + if os.IsNotExist(err) { + return nil, nil + } + return nil, err + } + + var info LeaderLockInfo + if err := json.Unmarshal(data, &info); err != nil { + return nil, err + } + + return &info, nil +} + +// healthCheckLoop periodically checks health and updates timestamps +func (e *Election) healthCheckLoop() { + defer e.wg.Done() + + ticker := time.NewTicker(time.Duration(e.config.HealthCheckIntervalMS) * time.Millisecond) + defer ticker.Stop() + + for { + select { + case <-e.stopChan: + return + case <-ticker.C: + e.performHealthCheck() + } + } +} + +// performHealthCheck checks health based on current role +func (e *Election) performHealthCheck() { + switch e.Role() { + case RoleLeader: + // Update timestamp in leader info + e.updateLeaderTimestamp() + + // Check if Redis is still running + if !e.redisManager.IsRunning() { + log.Println("[Election] Redis not running, attempting restart...") + if err := e.redisManager.Start(); err != nil { + log.Printf("[Election] Failed to restart Redis: %v", err) + } + } + + case RoleFollower: + // Re-read leader info in case it changed + info, err := e.readLeaderInfo() + if err != nil { + log.Printf("[Election] Failed to read leader info: %v", err) + return + } + + if info != nil { + e.mu.Lock() + e.leaderInfo = info + e.mu.Unlock() + } + } +} + +// updateLeaderTimestamp updates the timestamp in leader info +func (e *Election) updateLeaderTimestamp() { + e.mu.Lock() + if e.leaderInfo != nil { + e.leaderInfo.Timestamp = time.Now().UnixMilli() + info := *e.leaderInfo + e.mu.Unlock() + + if err := e.writeLeaderInfo(&info); err != nil { + log.Printf("[Election] Failed to update timestamp: %v", err) + } + } else { + e.mu.Unlock() + } +} + +// getLocalIP returns the local IP address +func getLocalIP() string { + addrs, err := net.InterfaceAddrs() + if err != nil { + hostname, _ := os.Hostname() + return hostname + } + + for _, addr := range addrs { + if ipnet, ok := addr.(*net.IPNet); ok && !ipnet.IP.IsLoopback() { + if ipnet.IP.To4() != nil { + return ipnet.IP.String() + } + } + } + + hostname, _ := os.Hostname() + return hostname +} diff --git a/internal/leader/redis.go b/internal/leader/redis.go new file mode 100644 index 0000000..40899fe --- /dev/null +++ b/internal/leader/redis.go @@ -0,0 +1,179 @@ +package leader + +import ( + "context" + "fmt" + "log" + "os" + "os/exec" + "sync" + "syscall" + "time" + + "github.com/metazla/meta-core/internal/config" + "github.com/redis/go-redis/v9" +) + +// RedisManager handles spawning and managing the Redis process +type RedisManager struct { + config *config.Config + cmd *exec.Cmd + running bool + mu sync.RWMutex +} + +// NewRedisManager creates a new Redis manager +func NewRedisManager(cfg *config.Config) *RedisManager { + return &RedisManager{ + config: cfg, + } +} + +// Start spawns the Redis server process +func (rm *RedisManager) Start() error { + rm.mu.Lock() + defer rm.mu.Unlock() + + if rm.running { + return nil + } + + // Ensure Redis data directory exists + dataDir := rm.config.RedisDataDir() + if err := os.MkdirAll(dataDir, 0755); err != nil { + return fmt.Errorf("failed to create Redis data directory: %w", err) + } + + log.Printf("[Redis] Spawning Redis on port %d...", rm.config.RedisPort) + + // Build Redis arguments matching TypeScript implementation + args := []string{ + "--port", fmt.Sprintf("%d", rm.config.RedisPort), + "--bind", "0.0.0.0", + "--dir", dataDir, + "--appendonly", "yes", + "--appendfilename", "appendonly.aof", + "--dbfilename", "dump.rdb", + "--save", "60", "1", // Save after 60 seconds if at least 1 key changed + "--loglevel", "warning", + } + + rm.cmd = exec.Command("redis-server", args...) + rm.cmd.Stdout = os.Stdout + rm.cmd.Stderr = os.Stderr + + if err := rm.cmd.Start(); err != nil { + return fmt.Errorf("failed to start Redis: %w", err) + } + + rm.running = true + log.Printf("[Redis] Started with PID %d", rm.cmd.Process.Pid) + + // Monitor the process in background + go rm.monitorProcess() + + return nil +} + +// Stop gracefully stops the Redis server +func (rm *RedisManager) Stop() error { + rm.mu.Lock() + defer rm.mu.Unlock() + + if !rm.running || rm.cmd == nil || rm.cmd.Process == nil { + return nil + } + + log.Println("[Redis] Stopping Redis...") + + // Try graceful shutdown first (SIGTERM) + if err := rm.cmd.Process.Signal(syscall.SIGTERM); err != nil { + log.Printf("[Redis] Failed to send SIGTERM: %v", err) + } + + // Wait up to 10 seconds for graceful shutdown + done := make(chan error, 1) + go func() { + done <- rm.cmd.Wait() + }() + + select { + case <-done: + log.Println("[Redis] Stopped gracefully") + case <-time.After(10 * time.Second): + log.Println("[Redis] Timeout, sending SIGKILL...") + rm.cmd.Process.Kill() + <-done + } + + rm.running = false + rm.cmd = nil + return nil +} + +// IsRunning returns true if Redis is currently running +func (rm *RedisManager) IsRunning() bool { + rm.mu.RLock() + defer rm.mu.RUnlock() + return rm.running +} + +// WaitForReady waits for Redis to be ready to accept connections +func (rm *RedisManager) WaitForReady(timeout time.Duration) error { + log.Println("[Redis] Waiting for Redis to be ready...") + + ctx, cancel := context.WithTimeout(context.Background(), timeout) + defer cancel() + + client := redis.NewClient(&redis.Options{ + Addr: fmt.Sprintf("localhost:%d", rm.config.RedisPort), + }) + defer client.Close() + + ticker := time.NewTicker(100 * time.Millisecond) + defer ticker.Stop() + + for { + select { + case <-ctx.Done(): + return fmt.Errorf("timeout waiting for Redis") + case <-ticker.C: + if err := client.Ping(ctx).Err(); err == nil { + log.Println("[Redis] Ready to accept connections") + return nil + } + } + } +} + +// Ping checks if Redis is responding +func (rm *RedisManager) Ping() error { + ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second) + defer cancel() + + client := redis.NewClient(&redis.Options{ + Addr: fmt.Sprintf("localhost:%d", rm.config.RedisPort), + }) + defer client.Close() + + return client.Ping(ctx).Err() +} + +// monitorProcess monitors the Redis process and updates state when it exits +func (rm *RedisManager) monitorProcess() { + if rm.cmd == nil { + return + } + + err := rm.cmd.Wait() + + rm.mu.Lock() + rm.running = false + rm.mu.Unlock() + + if err != nil { + log.Printf("[Redis] Process exited with error: %v", err) + } else { + log.Println("[Redis] Process exited normally") + } +} diff --git a/internal/storage/client.go b/internal/storage/client.go new file mode 100644 index 0000000..9caf941 --- /dev/null +++ b/internal/storage/client.go @@ -0,0 +1,549 @@ +package storage + +import ( + "context" + "fmt" + "log" + "strings" + "sync" + "time" + + "github.com/redis/go-redis/v9" +) + +// Client wraps Redis operations for metadata storage +type Client struct { + client *redis.Client + prefix string + connected bool + mu sync.RWMutex +} + +// NewClient creates a new storage client +func NewClient(prefix string) *Client { + return &Client{ + prefix: prefix, + } +} + +// Connect connects to Redis at the given URL +func (c *Client) Connect(url string) error { + c.mu.Lock() + defer c.mu.Unlock() + + // Parse Redis URL (redis://host:port) + addr := strings.TrimPrefix(url, "redis://") + + c.client = redis.NewClient(&redis.Options{ + Addr: addr, + DialTimeout: 5 * time.Second, + ReadTimeout: 30 * time.Second, + WriteTimeout: 30 * time.Second, + PoolSize: 10, + }) + + // Test connection + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + + if err := c.client.Ping(ctx).Err(); err != nil { + return fmt.Errorf("failed to connect to Redis: %w", err) + } + + c.connected = true + log.Printf("[Storage] Connected to Redis at %s", addr) + return nil +} + +// Close closes the Redis connection +func (c *Client) Close() error { + c.mu.Lock() + defer c.mu.Unlock() + + if c.client != nil { + c.connected = false + return c.client.Close() + } + return nil +} + +// IsConnected returns true if connected to Redis +func (c *Client) IsConnected() bool { + c.mu.RLock() + defer c.mu.RUnlock() + return c.connected +} + +// Health checks if Redis is responding +func (c *Client) Health() bool { + c.mu.RLock() + defer c.mu.RUnlock() + + if c.client == nil || !c.connected { + return false + } + + ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second) + defer cancel() + + return c.client.Ping(ctx).Err() == nil +} + +// buildKey constructs the full key with prefix +func (c *Client) buildKey(key string) string { + if c.prefix != "" { + return c.prefix + key + } + return key +} + +// buildFilePrefix constructs the prefix for a file's metadata keys +func (c *Client) buildFilePrefix(hashID string) string { + return fmt.Sprintf("/file/%s", hashID) +} + +// GetMetadataFlat retrieves all metadata for a file as a flat map +// Key format: /file/{hashId}/{property} -> value +func (c *Client) GetMetadataFlat(hashID string) (map[string]string, error) { + c.mu.RLock() + defer c.mu.RUnlock() + + if c.client == nil { + return nil, fmt.Errorf("not connected") + } + + ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second) + defer cancel() + + prefix := c.buildKey(c.buildFilePrefix(hashID)) + result := make(map[string]string) + + // Use SCAN to find all keys with this prefix + var cursor uint64 + for { + keys, nextCursor, err := c.client.Scan(ctx, cursor, prefix+"/*", 1000).Result() + if err != nil { + return nil, fmt.Errorf("scan failed: %w", err) + } + + if len(keys) > 0 { + // Get values for all keys + values, err := c.client.MGet(ctx, keys...).Result() + if err != nil { + return nil, fmt.Errorf("mget failed: %w", err) + } + + for i, key := range keys { + if values[i] != nil { + // Strip prefix to get property path + propPath := strings.TrimPrefix(key, prefix+"/") + result[propPath] = values[i].(string) + } + } + } + + cursor = nextCursor + if cursor == 0 { + break + } + } + + if len(result) == 0 { + return nil, nil + } + + return result, nil +} + +// SetMetadataFlat stores metadata for a file as flat key-value pairs +func (c *Client) SetMetadataFlat(hashID string, metadata map[string]string) error { + c.mu.RLock() + defer c.mu.RUnlock() + + if c.client == nil { + return fmt.Errorf("not connected") + } + + if len(metadata) == 0 { + return nil + } + + ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second) + defer cancel() + + prefix := c.buildKey(c.buildFilePrefix(hashID)) + + // Use pipeline for batch operations + pipe := c.client.Pipeline() + for prop, value := range metadata { + key := prefix + "/" + prop + pipe.Set(ctx, key, value, 0) + } + + _, err := pipe.Exec(ctx) + return err +} + +// GetAllHashIDs returns all unique file hash IDs stored +func (c *Client) GetAllHashIDs() ([]string, error) { + c.mu.RLock() + defer c.mu.RUnlock() + + if c.client == nil { + return nil, fmt.Errorf("not connected") + } + + ctx, cancel := context.WithTimeout(context.Background(), 60*time.Second) + defer cancel() + + prefix := c.buildKey("/file/") + hashSet := make(map[string]bool) + + // Scan for all file keys + var cursor uint64 + for { + keys, nextCursor, err := c.client.Scan(ctx, cursor, prefix+"*", 1000).Result() + if err != nil { + return nil, fmt.Errorf("scan failed: %w", err) + } + + for _, key := range keys { + // Extract hashID from key /file/{hashID}/property + stripped := strings.TrimPrefix(key, prefix) + parts := strings.SplitN(stripped, "/", 2) + if len(parts) > 0 && parts[0] != "" { + hashSet[parts[0]] = true + } + } + + cursor = nextCursor + if cursor == 0 { + break + } + } + + // Convert set to slice + result := make([]string, 0, len(hashSet)) + for hashID := range hashSet { + result = append(result, hashID) + } + + return result, nil +} + +// GetProperty retrieves a single property value +func (c *Client) GetProperty(hashID, property string) (string, error) { + c.mu.RLock() + defer c.mu.RUnlock() + + if c.client == nil { + return "", fmt.Errorf("not connected") + } + + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + + key := c.buildKey(c.buildFilePrefix(hashID) + "/" + property) + result, err := c.client.Get(ctx, key).Result() + if err == redis.Nil { + return "", nil + } + return result, err +} + +// SetProperty sets a single property value +func (c *Client) SetProperty(hashID, property, value string) error { + c.mu.RLock() + defer c.mu.RUnlock() + + if c.client == nil { + return fmt.Errorf("not connected") + } + + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + + key := c.buildKey(c.buildFilePrefix(hashID) + "/" + property) + return c.client.Set(ctx, key, value, 0).Err() +} + +// DeleteMetadata deletes all metadata for a file +func (c *Client) DeleteMetadata(hashID string) (int64, error) { + c.mu.RLock() + defer c.mu.RUnlock() + + if c.client == nil { + return 0, fmt.Errorf("not connected") + } + + ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second) + defer cancel() + + prefix := c.buildKey(c.buildFilePrefix(hashID)) + var deleted int64 + + // Scan and delete all keys with this prefix + var cursor uint64 + for { + keys, nextCursor, err := c.client.Scan(ctx, cursor, prefix+"/*", 1000).Result() + if err != nil { + return deleted, fmt.Errorf("scan failed: %w", err) + } + + if len(keys) > 0 { + count, err := c.client.Del(ctx, keys...).Result() + if err != nil { + return deleted, fmt.Errorf("del failed: %w", err) + } + deleted += count + } + + cursor = nextCursor + if cursor == 0 { + break + } + } + + return deleted, nil +} + +// CountFiles returns the number of unique files stored +func (c *Client) CountFiles() (int, error) { + hashIDs, err := c.GetAllHashIDs() + if err != nil { + return 0, err + } + return len(hashIDs), nil +} + +// LookupPathByCID searches all file metadata for a matching CID +// Returns the file path if found in poster/posterPath or backdrop/backdropPath +func (c *Client) LookupPathByCID(cid string) (string, error) { + c.mu.RLock() + defer c.mu.RUnlock() + + if c.client == nil { + return "", fmt.Errorf("not connected") + } + + if cid == "" { + return "", nil + } + + ctx, cancel := context.WithTimeout(context.Background(), 60*time.Second) + defer cancel() + + // Get all hash IDs first + hashIDs, err := c.getAllHashIDsInternal(ctx) + if err != nil { + return "", err + } + + // For each hash, check poster and backdrop CIDs + for _, hashID := range hashIDs { + prefix := c.buildKey(c.buildFilePrefix(hashID)) + + // Check poster + posterCID, err := c.client.Get(ctx, prefix+"/poster").Result() + if err == nil && posterCID == cid { + posterPath, err := c.client.Get(ctx, prefix+"/posterPath").Result() + if err == nil && posterPath != "" { + return posterPath, nil + } + } + + // Check backdrop + backdropCID, err := c.client.Get(ctx, prefix+"/backdrop").Result() + if err == nil && backdropCID == cid { + backdropPath, err := c.client.Get(ctx, prefix+"/backdropPath").Result() + if err == nil && backdropPath != "" { + return backdropPath, nil + } + } + } + + return "", nil +} + +// getAllHashIDsInternal is an internal version that doesn't acquire locks +// (caller must hold the lock) +func (c *Client) getAllHashIDsInternal(ctx context.Context) ([]string, error) { + prefix := c.buildKey("/file/") + hashSet := make(map[string]bool) + + var cursor uint64 + for { + keys, nextCursor, err := c.client.Scan(ctx, cursor, prefix+"*", 1000).Result() + if err != nil { + return nil, fmt.Errorf("scan failed: %w", err) + } + + for _, key := range keys { + stripped := strings.TrimPrefix(key, prefix) + parts := strings.SplitN(stripped, "/", 2) + if len(parts) > 0 && parts[0] != "" { + hashSet[parts[0]] = true + } + } + + cursor = nextCursor + if cursor == 0 { + break + } + } + + result := make([]string, 0, len(hashSet)) + for hashID := range hashSet { + result = append(result, hashID) + } + + return result, nil +} + +// MergeMetadataFlat merges new metadata into existing (PATCH semantics) +// New keys are added, existing keys are updated, missing keys are NOT deleted +func (c *Client) MergeMetadataFlat(hashID string, metadata map[string]string) (int, error) { + c.mu.RLock() + defer c.mu.RUnlock() + + if c.client == nil { + return 0, fmt.Errorf("not connected") + } + + if len(metadata) == 0 { + return 0, nil + } + + ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second) + defer cancel() + + prefix := c.buildKey(c.buildFilePrefix(hashID)) + + // Use pipeline for batch operations + pipe := c.client.Pipeline() + for prop, value := range metadata { + key := prefix + "/" + prop + pipe.Set(ctx, key, value, 0) + } + + _, err := pipe.Exec(ctx) + if err != nil { + return 0, err + } + + return len(metadata), nil +} + +// DeleteProperty deletes a single property +func (c *Client) DeleteProperty(hashID, property string) error { + c.mu.RLock() + defer c.mu.RUnlock() + + if c.client == nil { + return fmt.Errorf("not connected") + } + + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + + key := c.buildKey(c.buildFilePrefix(hashID) + "/" + property) + return c.client.Del(ctx, key).Err() +} + +// AddToSet adds a value to a set-type field (stored as pipe-delimited string) +// Returns true if value was added, false if it already existed +func (c *Client) AddToSet(hashID, property, value string) (bool, error) { + c.mu.Lock() + defer c.mu.Unlock() + + if c.client == nil { + return false, fmt.Errorf("not connected") + } + + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + + key := c.buildKey(c.buildFilePrefix(hashID) + "/" + property) + + // Get current value + current, err := c.client.Get(ctx, key).Result() + if err != nil && err != redis.Nil { + return false, err + } + + // Parse existing values (pipe-delimited) + var values []string + if current != "" { + values = strings.Split(current, "|") + } + + // Check if value already exists + for _, v := range values { + if v == value { + return false, nil // Already exists + } + } + + // Add new value + values = append(values, value) + newValue := strings.Join(values, "|") + + // Save back + err = c.client.Set(ctx, key, newValue, 0).Err() + if err != nil { + return false, err + } + + return true, nil +} + +// RemoveFromSet removes a value from a set-type field +// Returns true if value was removed, false if it didn't exist +func (c *Client) RemoveFromSet(hashID, property, value string) (bool, error) { + c.mu.Lock() + defer c.mu.Unlock() + + if c.client == nil { + return false, fmt.Errorf("not connected") + } + + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + + key := c.buildKey(c.buildFilePrefix(hashID) + "/" + property) + + // Get current value + current, err := c.client.Get(ctx, key).Result() + if err == redis.Nil { + return false, nil // Key doesn't exist + } + if err != nil { + return false, err + } + + // Parse existing values + values := strings.Split(current, "|") + + // Find and remove value + found := false + newValues := make([]string, 0, len(values)) + for _, v := range values { + if v == value { + found = true + } else { + newValues = append(newValues, v) + } + } + + if !found { + return false, nil + } + + // Save back (or delete if empty) + if len(newValues) == 0 { + return true, c.client.Del(ctx, key).Err() + } + + newValue := strings.Join(newValues, "|") + return true, c.client.Set(ctx, key, newValue, 0).Err() +} diff --git a/test/api_test.go b/test/api_test.go new file mode 100644 index 0000000..434ce3c --- /dev/null +++ b/test/api_test.go @@ -0,0 +1,174 @@ +package test + +import ( + "encoding/json" + "net/http" + "net/http/httptest" + "testing" + + "github.com/gorilla/mux" +) + +// MockHealthResponse represents the expected health response structure +type MockHealthResponse struct { + Status string `json:"status"` + Role string `json:"role"` + Redis bool `json:"redis"` + Timestamp string `json:"timestamp"` +} + +func TestHealthResponseFormat(t *testing.T) { + // Test that the health response has the expected format + response := MockHealthResponse{ + Status: "ok", + Role: "leader", + Redis: true, + Timestamp: "2024-01-01T00:00:00Z", + } + + data, err := json.Marshal(response) + if err != nil { + t.Fatalf("Failed to marshal: %v", err) + } + + var parsed map[string]interface{} + if err := json.Unmarshal(data, &parsed); err != nil { + t.Fatalf("Failed to unmarshal: %v", err) + } + + // Check required fields + requiredFields := []string{"status", "role", "redis", "timestamp"} + for _, field := range requiredFields { + if _, exists := parsed[field]; !exists { + t.Errorf("Missing required field: %s", field) + } + } +} + +func TestRouterSetup(t *testing.T) { + router := mux.NewRouter() + + // Register routes (simulating what the server does) + routes := []struct { + path string + method string + }{ + {"/health", "GET"}, + {"/status", "GET"}, + {"/leader", "GET"}, + {"/role", "GET"}, + {"/meta/{hash}", "GET"}, + {"/meta/{hash}", "PUT"}, + {"/meta/{hash}", "DELETE"}, + {"/meta", "GET"}, + {"/data/{hash}/path", "GET"}, + {"/data/{hash}", "HEAD"}, + {"/services", "GET"}, + {"/services/{name}", "GET"}, + } + + // Register dummy handlers + for _, route := range routes { + path := route.path + router.HandleFunc(path, func(w http.ResponseWriter, r *http.Request) { + w.WriteHeader(http.StatusOK) + }).Methods(route.method) + } + + // Test that routes are registered + testCases := []struct { + method string + path string + }{ + {"GET", "/health"}, + {"GET", "/status"}, + {"GET", "/leader"}, + {"GET", "/role"}, + {"GET", "/meta/testhash123"}, + {"PUT", "/meta/testhash123"}, + {"DELETE", "/meta/testhash123"}, + {"GET", "/meta"}, + {"GET", "/data/testhash123/path"}, + {"HEAD", "/data/testhash123"}, + {"GET", "/services"}, + {"GET", "/services/meta-sort"}, + } + + for _, tc := range testCases { + req := httptest.NewRequest(tc.method, tc.path, nil) + rr := httptest.NewRecorder() + + router.ServeHTTP(rr, req) + + if rr.Code != http.StatusOK { + t.Errorf("Route %s %s returned status %d, expected 200", tc.method, tc.path, rr.Code) + } + } +} + +func TestCORSHeaders(t *testing.T) { + // Create a simple handler + handler := http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.WriteHeader(http.StatusOK) + }) + + // Wrap with CORS middleware + corsHandler := func(next http.Handler) http.Handler { + return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.Header().Set("Access-Control-Allow-Origin", "*") + w.Header().Set("Access-Control-Allow-Methods", "GET, POST, PUT, DELETE, OPTIONS") + w.Header().Set("Access-Control-Allow-Headers", "Content-Type") + + if r.Method == "OPTIONS" { + w.WriteHeader(http.StatusOK) + return + } + + next.ServeHTTP(w, r) + }) + } + + req := httptest.NewRequest("GET", "/health", nil) + rr := httptest.NewRecorder() + + corsHandler(handler).ServeHTTP(rr, req) + + // Check CORS headers + if rr.Header().Get("Access-Control-Allow-Origin") != "*" { + t.Error("Missing or incorrect Access-Control-Allow-Origin header") + } + + if rr.Header().Get("Access-Control-Allow-Methods") == "" { + t.Error("Missing Access-Control-Allow-Methods header") + } +} + +func TestOptionsRequest(t *testing.T) { + corsHandler := func(next http.Handler) http.Handler { + return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.Header().Set("Access-Control-Allow-Origin", "*") + w.Header().Set("Access-Control-Allow-Methods", "GET, POST, PUT, DELETE, OPTIONS") + w.Header().Set("Access-Control-Allow-Headers", "Content-Type") + + if r.Method == "OPTIONS" { + w.WriteHeader(http.StatusOK) + return + } + + next.ServeHTTP(w, r) + }) + } + + handler := http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.WriteHeader(http.StatusOK) + }) + + req := httptest.NewRequest("OPTIONS", "/meta/test", nil) + rr := httptest.NewRecorder() + + corsHandler(handler).ServeHTTP(rr, req) + + if rr.Code != http.StatusOK { + t.Errorf("OPTIONS request returned status %d, expected 200", rr.Code) + } +} diff --git a/test/config_test.go b/test/config_test.go new file mode 100644 index 0000000..a75b115 --- /dev/null +++ b/test/config_test.go @@ -0,0 +1,91 @@ +package test + +import ( + "os" + "testing" + + "github.com/metazla/meta-core/internal/config" +) + +func TestConfigDefaults(t *testing.T) { + // Clear environment + os.Clearenv() + + cfg := config.Load() + + // Check defaults + if cfg.MetaCorePath != "/meta-core" { + t.Errorf("Expected MetaCorePath '/meta-core', got '%s'", cfg.MetaCorePath) + } + + if cfg.FilesPath != "/files" { + t.Errorf("Expected FilesPath '/files', got '%s'", cfg.FilesPath) + } + + if cfg.ServiceName != "meta-core" { + t.Errorf("Expected ServiceName 'meta-core', got '%s'", cfg.ServiceName) + } + + if cfg.HTTPPort != 9000 { + t.Errorf("Expected HTTPPort 9000, got %d", cfg.HTTPPort) + } + + if cfg.RedisPort != 6379 { + t.Errorf("Expected RedisPort 6379, got %d", cfg.RedisPort) + } +} + +func TestConfigFromEnv(t *testing.T) { + // Set environment variables + os.Setenv("META_CORE_PATH", "/custom/meta-core") + os.Setenv("FILES_PATH", "/custom/files") + os.Setenv("SERVICE_NAME", "test-service") + os.Setenv("META_CORE_HTTP_PORT", "8080") + os.Setenv("REDIS_PORT", "6380") + defer os.Clearenv() + + cfg := config.Load() + + if cfg.MetaCorePath != "/custom/meta-core" { + t.Errorf("Expected MetaCorePath '/custom/meta-core', got '%s'", cfg.MetaCorePath) + } + + if cfg.FilesPath != "/custom/files" { + t.Errorf("Expected FilesPath '/custom/files', got '%s'", cfg.FilesPath) + } + + if cfg.ServiceName != "test-service" { + t.Errorf("Expected ServiceName 'test-service', got '%s'", cfg.ServiceName) + } + + if cfg.HTTPPort != 8080 { + t.Errorf("Expected HTTPPort 8080, got %d", cfg.HTTPPort) + } + + if cfg.RedisPort != 6380 { + t.Errorf("Expected RedisPort 6380, got %d", cfg.RedisPort) + } +} + +func TestConfigPaths(t *testing.T) { + os.Setenv("META_CORE_PATH", "/test") + defer os.Clearenv() + + cfg := config.Load() + + if cfg.LockFilePath() != "/test/locks/kv-leader.lock" { + t.Errorf("Unexpected LockFilePath: %s", cfg.LockFilePath()) + } + + if cfg.InfoFilePath() != "/test/locks/kv-leader.info" { + t.Errorf("Unexpected InfoFilePath: %s", cfg.InfoFilePath()) + } + + if cfg.RedisDataDir() != "/test/db/redis" { + t.Errorf("Unexpected RedisDataDir: %s", cfg.RedisDataDir()) + } + + if cfg.ServicesDir() != "/test/services" { + t.Errorf("Unexpected ServicesDir: %s", cfg.ServicesDir()) + } +} diff --git a/test/election_test.go b/test/election_test.go new file mode 100644 index 0000000..740da8e --- /dev/null +++ b/test/election_test.go @@ -0,0 +1,144 @@ +package test + +import ( + "encoding/json" + "os" + "path/filepath" + "testing" + + "github.com/metazla/meta-core/internal/leader" +) + +func TestLeaderLockInfoJSON(t *testing.T) { + info := &leader.LeaderLockInfo{ + Host: "test-host", + API: "redis://10.0.0.1:6379", + HTTP: "http://10.0.0.1:8180", + BaseURL: "http://localhost:8180", + Timestamp: 1704067200000, + PID: 12345, + } + + // Marshal to JSON + data, err := json.MarshalIndent(info, "", " ") + if err != nil { + t.Fatalf("Failed to marshal: %v", err) + } + + // Verify JSON structure matches TypeScript format + var parsed map[string]interface{} + if err := json.Unmarshal(data, &parsed); err != nil { + t.Fatalf("Failed to unmarshal: %v", err) + } + + if parsed["host"] != "test-host" { + t.Errorf("Expected host 'test-host', got '%v'", parsed["host"]) + } + + if parsed["api"] != "redis://10.0.0.1:6379" { + t.Errorf("Expected api 'redis://10.0.0.1:6379', got '%v'", parsed["api"]) + } + + if parsed["http"] != "http://10.0.0.1:8180" { + t.Errorf("Expected http 'http://10.0.0.1:8180', got '%v'", parsed["http"]) + } + + if parsed["baseUrl"] != "http://localhost:8180" { + t.Errorf("Expected baseUrl 'http://localhost:8180', got '%v'", parsed["baseUrl"]) + } + + // Timestamp should be a number + if _, ok := parsed["timestamp"].(float64); !ok { + t.Errorf("Expected timestamp to be a number, got %T", parsed["timestamp"]) + } + + // PID should be a number + if _, ok := parsed["pid"].(float64); !ok { + t.Errorf("Expected pid to be a number, got %T", parsed["pid"]) + } +} + +func TestLeaderLockInfoJSONOmitEmpty(t *testing.T) { + info := &leader.LeaderLockInfo{ + Host: "test-host", + API: "redis://10.0.0.1:6379", + HTTP: "http://10.0.0.1:8180", + BaseURL: "", // Empty - should be omitted + Timestamp: 1704067200000, + PID: 12345, + } + + data, err := json.Marshal(info) + if err != nil { + t.Fatalf("Failed to marshal: %v", err) + } + + // baseUrl should not be present when empty + var parsed map[string]interface{} + if err := json.Unmarshal(data, &parsed); err != nil { + t.Fatalf("Failed to unmarshal: %v", err) + } + + if _, exists := parsed["baseUrl"]; exists { + t.Error("Expected baseUrl to be omitted when empty") + } +} + +func TestLeaderInfoFileFormat(t *testing.T) { + // Create a temp directory + tmpDir, err := os.MkdirTemp("", "meta-core-test") + if err != nil { + t.Fatalf("Failed to create temp dir: %v", err) + } + defer os.RemoveAll(tmpDir) + + info := &leader.LeaderLockInfo{ + Host: "test-host", + API: "redis://10.0.0.1:6379", + HTTP: "http://10.0.0.1:8180", + Timestamp: 1704067200000, + PID: 12345, + } + + // Write to file + infoPath := filepath.Join(tmpDir, "kv-leader.info") + data, err := json.MarshalIndent(info, "", " ") + if err != nil { + t.Fatalf("Failed to marshal: %v", err) + } + + if err := os.WriteFile(infoPath, data, 0644); err != nil { + t.Fatalf("Failed to write file: %v", err) + } + + // Read and parse + readData, err := os.ReadFile(infoPath) + if err != nil { + t.Fatalf("Failed to read file: %v", err) + } + + var parsed leader.LeaderLockInfo + if err := json.Unmarshal(readData, &parsed); err != nil { + t.Fatalf("Failed to unmarshal: %v", err) + } + + if parsed.Host != info.Host { + t.Errorf("Host mismatch: expected '%s', got '%s'", info.Host, parsed.Host) + } + + if parsed.API != info.API { + t.Errorf("API mismatch: expected '%s', got '%s'", info.API, parsed.API) + } + + if parsed.HTTP != info.HTTP { + t.Errorf("HTTP mismatch: expected '%s', got '%s'", info.HTTP, parsed.HTTP) + } + + if parsed.Timestamp != info.Timestamp { + t.Errorf("Timestamp mismatch: expected %d, got %d", info.Timestamp, parsed.Timestamp) + } + + if parsed.PID != info.PID { + t.Errorf("PID mismatch: expected %d, got %d", info.PID, parsed.PID) + } +}