TL;DR — Build a real-time user safety pipeline called safe. in Go that ingests user-generated content, runs it through parallel ML classifiers (toxicity, PII detection, spam), and takes action based on a configurable risk policy. This project demonstrates concurrent pipeline design, polyglot architecture, observability, and fault tolerance — the exact skills that separate staff engineers from the rest.

Introduction

Every platform that touches user-generated content needs a safety layer. Content moderation isn’t a nice-to-have — it’s a regulatory requirement, a trust signal, and a business imperative. Yet most portfolio projects stop at CRUD apps and todo lists. If you want to signal that you can design systems that are actually deployed in production, you need a project that deals with concurrency, real-time decisioning, and integration with external services.

safe. is that project. It’s a content safety pipeline — a service that accepts user content (text messages, comments, posts), runs it through multiple parallel classifiers, aggregates risk scores against a configurable policy, and decides whether to allow, flag, or block the content. Along the way, you’ll build skills in Go concurrency patterns, gRPC service design, structured logging, distributed tracing, and ML service integration.

The best part: this is not hypothetical. Companies like Discord, Reddit, and OpenAI run variations of exactly this architecture at scale. By building safe., you’re not just building a project — you’re internalizing the patterns that power real-world safety infrastructure.

Why This Project Stands Out on a CV

Hiring managers and staff-level engineers scan portfolios for specific signals. safe. hits nearly all of them:

  • Concurrency and parallelism. The pipeline processes content through multiple classifiers simultaneously using Go’s goroutines and channels. This is the same pattern used in production systems like Kafka consumers and cloud load balancers.
  • Polyglot architecture. The pipeline engine is Go; the ML classifiers are Python microservices exposed via gRPC. You demonstrate you can design systems that span language boundaries — a skill almost every senior role demands.
  • Real-time decisioning. Unlike batch-processing projects, safe. makes latency-sensitive decisions under a configurable deadline. This signals you understand SLIs, SLOs, and the tradeoff between accuracy and speed.
  • Observability. Structured JSON logging, OpenTelemetry traces, and Prometheus metrics are wired in from day one. This is the difference between “it works on my machine” and “I can diagnose it at 3 AM in production.”
  • Fault tolerance. Circuit breakers, retry policies, and graceful degradation mean the system keeps running even when a classifier service goes down. This is the kind of thinking that separates juniors from seniors.
  • Policy-driven configuration. The risk thresholds and actions are configurable via YAML, not hardcoded. This signals you understand that production systems must adapt without redeployment.

The roles this signals: Backend Engineer, Platform Engineer, Trust & Safety Engineer, Site Reliability Engineer, and Senior Software Engineer — essentially any role where system design and reliability matter.

Architecture Overview

safe. is composed of five distinct services that communicate over gRPC. Here’s the high-level view:

┌──────────────┐     ┌──────────────────┐     ┌─────────────────────┐
│  Client /    │────▶│  API Gateway     │────▶│  Safety Pipeline    │
│  Frontend    │     │  (HTTP → gRPC)   │     │  Orchestrator       │
└──────────────┘     └──────────────────┘     └─────────┬───────────┘
                                                          │
                    ┌─────────────────────────────────────┼──────────────┐
                    │                                     │              │
            ┌───────▼───────┐                   ┌────────▼──────┐ ┌─────▼──────┐
            │  Toxicity     │                   │  PII Detector │ │  Spam      │
            │  Classifier   │                   │  Service      │ │  Classifier  │
            │  (Python)     │                   │  (Python)     │ │  (Python)   │
            └───────────────┘                   └───────────────┘ └─────────────┘
                                                          │
                    ┌─────────────────────────────────────┼──────────────┐
                    │                                     │              │
            ┌───────▼───────────┐                 ┌────────▼────────┐ ┌──▼─────────┐
            │  Policy Engine   │                 │  Action Executor│ │  Metrics /  │
            │  (Risk Scorer)   │                 │  (Block/Flag/   │ │  Observability│
            │                  │                 │   Allow)        │ │  (Prometheus│
            └─────────────────┘                 └─────────────────┘ │  / Grafana) │
                                                                     │
                                                          ┌────────▼────────┐
                                                          │  Persistent     │
                                                          │  Store (SQLite) │
                                                          └─────────────────┘

Here’s what each component does:

  1. API Gateway — Accepts HTTP requests, translates them to gRPC, and forwards content to the pipeline orchestrator. Written in Go using gin-gonic for HTTP and grpc-gateway for proto-to-HTTP translation.

  2. Pipeline Orchestrator — The heart of safe.. Receives a content item, fans out classification tasks to all classifiers concurrently via goroutines and channels, collects results, and passes aggregated scores to the policy engine. Uses a context.WithTimeout to enforce a global deadline.

  3. Classifier Services — Independent Python microservices (one per category: toxicity, PII, spam). Each loads a small ONNX or HuggingFace model and exposes a gRPC Classify method returning a risk score (0.0–1.0) and a list of flagged categories.

  4. Policy Engine — A configurable rules engine that maps aggregated risk scores to actions. Defined in YAML:

    rules:
  - name: "high_toxicity"
    condition: "toxicity_score > 0.85"
    action: "block"
  - name: "moderate_pii"
    condition: "pii_score > 0.5 and toxicity_score < 0.3"
    action: "flag"
  - name: "allow"
    condition: "default"
    action: "allow"

  5. Action Executor — Takes the policy engine’s decision and persists it to a SQLite store, emits an event to a local message queue (or NATS for extension), and returns the decision to the caller.

  6. Observability Stack — Every request gets a trace ID propagated through all services. Structured logs go to stdout in JSON format. Prometheus metrics track request latency, classifier error rates, and decision distribution. Grafana dashboards visualize them.

Building It Step by Step

We’ll build the core pipeline orchestrator in Go. The classifier services are Python gRPC servers. Here’s the full implementation.

Step 1: Define the Protocol Buffers

Create proto/safety.proto:

syntax = "proto3";

package safety;

service Classifier {
  rpc Classify(ClassifyRequest) returns (ClassifyResponse);
}

message ClassifyRequest {
  string content = 1;
  string request_id = 2;
}

message ClassifyResponse {
  float score = 1;
  repeated string categories = 2;
  string classifier_name = 3;
  bool healthy = 4;
}

message SafetyCheckRequest {
  string content = 1;
  string user_id = 2;
  string request_id = 3;
}

message SafetyCheckResponse {
  string decision = 1;        // "allow", "flag", "block"
  float overall_risk = 2;
  map<string, float> scores = 3;
  repeated string flagged_categories = 4;
  int64 processing_time_ms = 5;
}

Generate Go stubs:

protoc --go_out=. --go-grpc_out=. --grpc-gateway_out=. proto/safety.proto

Step 2: Build the Pipeline Orchestrator (Go)

The orchestrator is where the real systems engineering lives. It fans out classification tasks concurrently and collects results with a deadline:

package main

import (
	"context"
	"fmt"
	"sync"
	"time"

	pb "safe/proto"
	"google.golang.org/grpc"
	"google.golang.org/grpc/credentials/insecure"
)

type ClassifierClient struct {
	client pb.ClassifierClient
	conn   *grpc.ClientConn
	name   string
}

func dialClassifier(addr string) (*ClassifierClient, error) {
	conn, err := grpc.Dial(addr, grpc.WithTransportCredentials(insecure.NewCredentials()))
	if err != nil {
		return nil, fmt.Errorf("dialing %s: %w", addr, err)
	}
	return &ClassifierClient{
		client: pb.NewClassifierClient(conn),
		conn:   conn,
	}, nil
}

// classifyAsync runs a single classifier in a goroutine and sends the
// result (or error) on the provided channel.
func classifyAsync(ctx context.Context, cc *ClassifierClient, content, reqID string, ch chan<- *pb.ClassifyResponse) {
	defer func() {
		if r := recover(); r != nil {
			ch <- &pb.ClassifyResponse{Healthy: false}
		}
	}()

	resp, err := cc.client.Classify(ctx, &pb.ClassifyRequest{
		Content:   content,
		RequestId: reqID,
	})
	if err != nil {
		ch <- &pb.ClassifyResponse{Healthy: false}
		return
	}
	ch <- resp
}

// RunPipeline is the core orchestration logic. It fans out to all
// classifiers concurrently, enforces a global deadline, and returns
// the aggregated results.
func RunPipeline(ctx context.Context, classifiers []*ClassifierClient, content, reqID string) (*pb.SafetyCheckResponse, error) {
	deadline := time.Now().Add(2 * time.Second)
	ctx, cancel := context.WithTimeout(ctx, deadline.Sub(time.Now()))
	defer cancel()

	results := make(chan *pb.ClassifyResponse, len(classifiers))
	var wg sync.WaitGroup

	// Fan-out: launch every classifier concurrently.
	for _, cc := range classifiers {
		wg.Add(1)
		go func(client *ClassifierClient) {
			defer wg.Done()
			classifyAsync(ctx, client, content, reqID, results)
		}(cc)
	}

	// Close the channel once all goroutines complete.
	go func() {
		wg.Wait()
		close(results)
	}()

	// Collect results with timeout awareness.
	scores := make(map[string]float32)
	var flaggedCategories []string
	var totalRisk float32
	var healthyCount int

	for resp := range results {
		if !resp.Healthy {
			continue // graceful degradation: skip failed classifiers
		}
		healthyCount++
		for _, cat := range resp.Categories {
			scores[cat] = resp.Score
			totalRisk += resp.Score
			flaggedCategories = append(flaggedCategories, cat)
		}
	}

	if healthyCount == 0 {
		// Fail open: if all classifiers are down, allow by default
		// but log a critical alert. This is the fault-tolerance choice.
		return &pb.SafetyCheckResponse{
			Decision:     "allow",
			OverallRisk:  0,
			Scores:       scores,
			FlaggedCategories: flaggedCategories,
		}, nil
	}

	avgRisk := totalRisk / float32(healthyCount)
	decision := evaluatePolicy(avgRisk, scores)

	return &pb.SafetyCheckResponse{
		Decision:          decision,
		OverallRisk:       avgRisk,
		Scores:            scores,
		FlaggedCategories: flaggedCategories,
	}, nil
}

func evaluatePolicy(avgRisk float32, scores map[string]float32) string {
	if scores["toxicity"] > 0.85 || avgRisk > 0.8 {
		return "block"
	}
	if scores["pii"] > 0.5 || avgRisk > 0.5 {
		return "flag"
	}
	return "allow"
}

Step 3: Build the HTTP API Gateway (Go)

package main

import (
	"encoding/json"
	"net/http"
	"time"

	"github.com/gin-gonic/gin"
	"github.com/google/uuid"
	"go.uber.org/zap" // structured logging
)

type SafetyHandler struct {
	pipeline *PipelineOrchestrator
	logger   *zap.Logger
}

func (h *SafetyHandler) CheckContent(c *gin.Context) {
	start := time.Now()
	reqID := uuid.New().String()

	var req struct {
		Content string `json:"content" binding:"required"`
		UserID  string `json:"user_id"`
	}
	if err := c.ShouldBindJSON(&req); err != nil {
		h.logger.Warn("invalid request", zap.String("req_id", reqID), zap.Error(err))
		c.JSON(http.StatusBadRequest, gin.H{"error": err.Error()})
		return
	}

	resp, err := h.pipeline.RunPipeline(c.Request.Context(), req.Content, reqID)
	if err != nil {
		h.logger.Error("pipeline failed", zap.String("req_id", reqID), zap.Error(err))
		c.JSON(http.StatusInternalServerError, gin.H{"error": "safety check failed"})
		return
	}

	latency := float64(time.Since(start).Milliseconds())
	h.logger.Info("safety check complete",
		zap.String("req_id", reqID),
		zap.String("decision", resp.Decision),
		zap.Float64("latency_ms", latency),
		zap.Float32("overall_risk", resp.OverallRisk),
	)

	c.JSON(http.StatusOK, gin.H{
		"request_id":     reqID,
		"decision":       resp.Decision,
		"overall_risk":   resp.OverallRisk,
		"scores":         resp.Scores,
		"flagged":        resp.FlaggedCategories,
		"processing_ms":  latency,
	})
}

Step 4: Build a Classifier Service (Python)

Each classifier is a standalone Python gRPC server. Here’s the toxicity classifier using a HuggingFace model:

import grpc
from concurrent import futures
import toxicity_pb2
import toxicity_pb2_grpc
import torch
from transformers import AutoTokenizer, AutoModelForSequenceClassification
import logging
import time

logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s')
logger = logging.getLogger("toxicity_classifier")

class ToxicityClassifier(toxicity_pb2_grpc.ClassifierServicer):
    def __init__(self, model_name="unitary/toxic-bert"):
        self.tokenizer = AutoTokenizer.from_pretrained(model_name)
        self.model = AutoModelForSequenceClassification.from_pretrained(model_name)
        self.model.eval()
        logger.info("Toxicity classifier loaded successfully")

    def Classify(self, request, context):
        start = time.time()
        inputs = self.tokenizer(request.content, return_tensors="pt", truncation=True, max_length=512)

        with torch.no_grad():
            outputs = self.model(**inputs)
            score = torch.sigmoid(outputs.logits[0][1]).item()

        categories = ["toxicity"] if score > 0.5 else []
        elapsed = (time.time() - start) * 1000

        logger.info(f"Classified request={request.request_id} score={score:.4f} latency={elapsed:.1f}ms")

        return toxicity_pb2.ClassifyResponse(
            score=score,
            categories=categories,
            classifier_name="toxicity",
            healthy=True
        )

def serve():
    server = grpc.server(futures.ThreadPoolExecutor(max_workers=4))
    toxicity_pb2_grpc.add_ClassifierServicer_to_server(ToxicityClassifier(), server)
    server.add_insecure_port("[::]:50051")
    server.start()
    logger.info("Toxicity classifier gRPC server listening on :50051")
    server.wait_for_termination()

if __name__ == "__main__":
    serve()

Step 5: Wire Up the Policy Engine

The policy engine reads the YAML config and evaluates rules in order. First, define the config schema:

// policy.go
package main

import (
	"os"
	"gopkg.in/yaml.v3"
)

type Rule struct {
	Name     string  `yaml:"name"`
	Condition string  `yaml:"condition"`
	Action   string  `yaml:"action"`
}

type PolicyConfig struct {
	Rules []Rule `yaml:"rules"`
	DefaultAction string `yaml:"default_action"`
}

func LoadPolicy(path string) (*PolicyConfig, error) {
	data, err := os.ReadFile(path)
	if err != nil {
		return nil, err
	}
	var cfg PolicyConfig
	if err := yaml.Unmarshal(data, &cfg); err != nil {
		return nil, err
	}
	return &cfg, nil
}

func (p *PolicyConfig) Evaluate(scores map[string]float32) string {
	for _, rule := range p.Rules {
		if rule.Condition == "default" {
			return rule.Action
		}
		// Simple expression evaluation — in production, use a proper
		// expression engine like github.com/Knetic/govaluate
		if evaluateCondition(rule.Condition, scores) {
			return rule.Action
		}
	}
	return p.DefaultAction
}

Running and Testing It

Local Setup

# 1. Clone and set up the project
git clone https://github.com/your-org/safe.git
cd safe

# 2. Generate protobuf stubs
make proto  # runs protoc with go plugins

# 3. Start classifier services (each in its own terminal)
python3 classifiers/toxicity_server.py &
python3 classifiers/pii_server.py &
python3 classifiers/spam_server.py &

# 4. Start the API gateway
go build -o safe ./cmd/safe
./safe --config config/policy.yaml --port :8080

# 5. Verify it's running
curl http://localhost:8080/health
# {"status":"ok","classifiers":3}

Sending a Request

curl -X POST http://localhost:8080/v1/safety/check \
  -H "Content-Type: application/json" \
  -d '{
    "content": "You are a terrible person and I hope you fail",
    "user_id": "user-1234"
  }'

Expected response:

{
  "request_id": "a1b2c3d4-e5f6-7890-abcd-ef1234567890",
  "decision": "block",
  "overall_risk": 0.87,
  "scores": {"toxicity": 0.92},
  "flagged": ["toxicity"],
  "processing_ms": 340.2
}

Testing with a Test Suite

// pipeline_test.go
package main

import (
	"context"
	"testing"
	"time"
)

func TestPipeline_AllowsSafeContent(t *testing.T) {
	classifiers := mockClassifiers(map[string]float32{
		"toxicity": 0.05,
		"pii":      0.01,
		"spam":     0.02,
	})

	resp, err := RunPipeline(context.Background(), classifiers, "Hello world!", "test-1")
	if err != nil {
		t.Fatalf("unexpected error: %v", err)
	}
	if resp.Decision != "allow" {
		t.Errorf("expected 'allow', got %q", resp.Decision)
	}
	if resp.OverallRisk > 0.2 {
		t.Errorf("expected low risk, got %f", resp.OverallRisk)
	}
}

func TestPipeline_BlocksHighToxicity(t *testing.T) {
	classifiers := mockClassifiers(map[string]float32{
		"toxicity": 0.95,
		"pii":      0.1,
		"spam":     0.0,
	})

	resp, err := RunPipeline(context.Background(), classifiers, "You suck!", "test-2")
	if err != nil {
		t.Fatalf("unexpected error: %v", err)
	}
	if resp.Decision != "block" {
		t.Errorf("expected 'block', got %q", resp.Decision)
	}
}

func TestPipeline_GracefulDegradation(t *testing.T) {
	// All classifiers report unhealthy — system should fail open
	classifiers := []*ClassifierClient{deadClassifier(), deadClassifier(), deadClassifier()}

	resp, err := RunPipeline(context.Background(), classifiers, "test content", "test-3")
	if err != nil {
		t.Fatalf("unexpected error: %v", err)
	}
	if resp.Decision != "allow" {
		t.Errorf("expected fail-open 'allow', got %q", resp.Decision)
	}
}

func TestPipeline_DeadlineEnforcement(t *testing.T) {
	slowClassifier := mockSlowClassifier(5 * time.Second) // simulates a hung service
	classifiers := []*ClassifierClient{slowClassifier}

	ctx, cancel := context.WithTimeout(context.Background(), 100*time.Millisecond)
	defer cancel()

	_, err := RunPipeline(ctx, classifiers, "test", "test-timeout")
	if err == nil {
		t.Error("expected deadline error, got nil")
	}
}

Run the suite:

go test ./... -v -race

The -race flag is critical — it detects data races in your concurrent pipeline, which is exactly the kind of bug that causes outages in production.

Running with Observability

# Start with Prometheus metrics endpoint
./safe --config config/policy.yaml --port :8080 --metrics :9090

# In another terminal, check metrics
curl http://localhost:9090/metrics | grep safety
# safety_requests_total{decision="block"} 42
# safety_request_duration_seconds_sum 0.34

Extending It: Your Roadmap to Senior-Level

This is where safe. transforms from a portfolio project into something that genuinely signals senior-level thinking. Each upgrade maps to a real production concern.

  1. Add Persistent Storage and Audit Logging. Replace the in-memory decision log with SQLite or PostgreSQL. Every safety decision must be persisted with the request ID, timestamp, content hash, and decision — this is non-negotiable for compliance (GDPR, CCPA, DSA). It matters because you cannot operate a safety system without an audit trail.

  2. Horizontal Scaling with a Message Queue. Replace the direct goroutine fan-out with NATS or Kafka. The orchestrator publishes classification tasks to a topic; each classifier is a consumer group member. This lets you scale classifiers independently and handle traffic spikes. It matters because production systems must handle load that exceeds a single process’s capacity.

  3. Distributed Tracing with OpenTelemetry. Instrument every service with OTel SDKs. Propagate trace context through gRPC metadata. Visualize the full request lifecycle in Jaeger. It matters because when a request takes 2 seconds instead of 200ms, you need to know exactly which service is the bottleneck.

  4. Circuit Breaker Pattern. Integrate sony/gobreaker into the orchestrator. If a classifier fails more than 5 times in 60 seconds, open the circuit and skip it for 30 seconds. Combine with a health-check endpoint that other orchestrators can poll. It matters because a single failing dependency should never cascade into a system-wide outage.

  5. Benchmarking and Performance Profiling. Use Go’s pprof and benchstat to benchmark the pipeline under load. Add a go test -bench suite that measures throughput (requests/second) and p99 latency as you add classifiers. Set up a CI pipeline that fails if p99 latency regresses by more than 10%. It matters because performance is a feature — a safety system that’s too slow becomes a denial-of-service vector.

  6. Model Versioning and A/B Testing. Add a model registry (MLflow or a simple S3-backed version store). Allow the orchestrator to route a percentage of traffic to a new classifier version and compare decisions. Track metrics like precision, recall, and false-positive rate per version. It matters because in safety systems, a model update that increases false positives by 2% can suppress legitimate speech at scale.

Key Takeaways

  • safe. demonstrates systems thinking, not just coding. The concurrent pipeline architecture, fault tolerance, and policy-driven design are the same patterns used in production at scale.
  • Concurrency is the differentiator. Go’s goroutines and channels let you fan out to multiple classifiers in parallel — this is the single most impressive technical element on a CV.
  • Observability is not optional. Structured logging, metrics, and tracing from day one signal that you’ve shipped systems before, not just built demos.
  • Graceful degradation is a senior-level decision. Choosing to fail open (allow content) when classifiers are down, rather than failing hard, shows you understand availability vs. safety tradeoffs.
  • Each extension maps to a real production problem. Persistence → compliance. Message queues → scalability. Circuit breakers → reliability. Model versioning → ML ops. These are not hypotheticals — they are the daily concerns of trust and safety teams at every major platform.

Further Reading

  1. The Log: What Every Software Engineer Should Know About Real-Time Data’s Unifying Abstraction — Jay Kreps’ foundational paper on the log abstraction. Understanding this is essential when you add Kafka/NATS to safe. for the message queue extension.

  2. gRPC Performance Best Practices — The official gRPC blog post on optimizing throughput and latency. Directly applicable when you benchmark the classifier communication layer.

  3. Designing Data-Intensive Applications by Martin Kleppmann — Chapters 4 (Encoding and Evolution), 7 (Consistency and Consensus), and 11 (Stream Processing) are the canonical references for everything from model versioning to pipeline state management.

  4. OpenTelemetry Specification — The canonical docs for distributed tracing and metrics. Required reading when you instrument safe. with OTel.

  5. The Circuit Breaker Pattern — Martin Fowler’s original description of the circuit breaker pattern. The theoretical foundation for the fault-tolerance extension.

  6. Content Moderation at Scale: Engineering Trust & Safety — OpenAI’s engineering blog on building content moderation systems at scale. Provides real-world context for the architecture decisions in safe..