Stream Framework

Go REST + gRPC framework for production services

This documentation is served locally at /docs for development reference.

Stream Framework is a production-ready Go service template that combines the speed of Fiber (FastHTTP) with OpenTelemetry distributed tracing, Sentry error monitoring, a pluggable queue system, and a clean module-based architecture.

Tech Stack

Layer	Technology
HTTP Server	Fiber v2 (FastHTTP)
gRPC	google.golang.org/grpc
Database ORM	GORM + PostgreSQL
Cache	Redis
Tracing	OpenTelemetry (OTLP)
Error Tracking	Sentry
Queue	Database (SKIP LOCKED) or RabbitMQ
Auth	JWT + Redis-backed sessions
Logging	Zap (file, S3, GCP, Graylog)

Architecture Overview

flowchart LR subgraph EXT[" Clients "] C1["REST Client\n(Browser / App)"] C2["gRPC Client"] end subgraph SFW[" Stream Framework "] subgraph TRANSPORT["Transport Layer"] HTTP["Fiber v2\nHTTP · WebSocket"] GRPC["gRPC Server"] end subgraph MW["Middleware"] MID["CORS · Logger · Recover\nJWT Auth · Permissions"] end subgraph MODULE["Module Layer"] direction TB CTL["Controller"] --> SVC["Service"] --> REPO["Repository"] end end subgraph INFRA[" Infrastructure "] PG[("PostgreSQL\nGORM")] RD[("Redis\nCache · Session")] QUE["Queue\nDB SKIP LOCKED\nor RabbitMQ"] FS["File Storage\nS3 · GCP · Local"] end subgraph OBS[" Observability "] OTEL["OpenTelemetry\nOTLP Traces"] SENT["Sentry\nErrors"] LOG["Zap Logger\nFile · S3 · GCP · Graylog"] end C1 -->|HTTP / WS| HTTP C2 -->|protobuf| GRPC HTTP --> MID GRPC --> MID MID --> CTL REPO --> PG REPO --> RD SVC --> QUE SVC --> FS SFW -.->|traces| OTEL SFW -.->|errors| SENT SFW -.->|logs| LOG

Development Life-cycle

Start → Finish: delivering a feature on Stream Framework

ทุก Feature ที่ส่ง Delivery ผ่านขั้นตอนมาตรฐาน 9 ขั้นตอนนี้ ตั้งแต่เปิด Branch จนถึง Merge เข้า develop

ขั้นตอนการพัฒนา

1. เปิด Feature Branch

แตก branch ใหม่จาก develop โดยใช้ชื่อ feature/<name> หรือ fix/<name>

git checkout develop && git pull
git checkout -b feature/my-feature

2. Scaffold Module (ถ้าเป็น Feature ใหม่)

ใช้ Code Generator สร้าง boilerplate ทั้งหมดในคราวเดียว

# สร้าง module ครบชุด (controller, service, repository, dto, resource, route, error)
./gen module <name>

# หรือสร้างเฉพาะ component ที่ต้องการ
./gen module <name> controller,service

# สร้างเฉพาะ GORM model
./gen model <name>

3. ออกแบบ Model & Migration

สร้าง GORM model ใน src/models/ และไฟล์ SQL migration ใน database/migrations/ ตามรูปแบบ golang-migrate

-- database/migrations/000010_create_items_table.up.sql
CREATE TABLE tbl_items (
    id         BIGSERIAL PRIMARY KEY,
    name       VARCHAR(255) NOT NULL,
    created_at TIMESTAMPTZ DEFAULT NOW(),
    updated_at TIMESTAMPTZ DEFAULT NOW()
);

4. Implement Repository

เขียน Repository ใน module file สำหรับ data-access logic ทั้งหมด Repository ต้องใช้ interface เพื่อรองรับ fake ใน test

5. Implement Service

เขียน business logic ใน Service รับ repository ผ่าน WithXxxRepositoryFactory option เพื่อให้ inject fake ได้ตอน test

6. Implement Controller

เขียน HTTP handler ใน Controller โดยเรียก Service และใช้ exception.HttpErrorResponseMapping สำหรับ error response ทุกกรณี

7. ลงทะเบียน Routes

เพิ่ม route ของ module ใน src/routes/http.go ภายใน HTTPRoutes() เท่านั้น

// src/routes/http.go
api.Mount("/items", item_module.NewRouter(db, redis))

8. เขียน Tests

เขียน unit test โดย inject fake repository ผ่าน WithXxxRepositoryFactory และเรียก setupXxxTestGlobals(t) เพื่อ init singleton ก่อนรัน test

go test ./src/modules/item_module/...

9. Regenerate Swagger & ทดสอบ Local

รัน swag init เพื่ออัปเดต Swagger docs และใช้ air สำหรับ live reload ระหว่างพัฒนา

swag fmt && swag init
air

อัปเดต .env.dev และ .env.example ทุกครั้งที่เพิ่ม environment variable ใหม่

ภาพรวมขั้นตอน

flowchart TD A(["🌿 git checkout -b feature/xxx"]) --> B["./gen module <name>"] B --> C["ออกแบบ Model\n+ SQL Migration"] C --> D["Implement\nRepository"] D --> E["Implement\nService"] E --> F["Implement\nController"] F --> G["Register Route\nsrc/routes/http.go"] G --> H["เขียน Tests\ngo test ./..."] H --> I["swag init\n+ air"] I --> J(["✅ Open Pull Request → develop"])

Requirements

Runtime

Dependency	Minimum Version	Notes
Go	1.24+	Required for latest language features
PostgreSQL	14+	Primary data store; SKIP LOCKED for queue
Redis	6+	Cache + session store
RabbitMQ	3.x	Optional; alternative queue driver

Development Tools

air — live reload: go install github.com/cosmtrek/air@latest
swag — Swagger generator: go install github.com/swaggo/swag/cmd/swag@latest
migrate — database migrations: golang-migrate
protoc — Protocol Buffer compiler (only if modifying gRPC definitions)

Installation

1. Clone the repository

git clone https://gitlab.innovationstream.app/stream/stream-framework.git myapp
cd myapp

2. Install Go dependencies

go mod tidy

3. Copy environment file

cp .env.example .env.dev
# Edit .env.dev with your local settings

4. Run database migrations

migrate -path database/migrations -database "postgres://user:pass@localhost/mydb?sslmode=disable" up

5. Start the server

air          # with live reload
# or
go run .     # without live reload

When ENV is unset, local, or development, the framework automatically loads .env.dev. In production, it loads .env.

stream-go CLI

stream-go scaffolds a brand-new project from this framework — the recommended way to start. It ships the whole template embedded in the binary, rewrites the module path, and toggles optional subsystems for you.

1. Install

The framework is a private module, so set GOPRIVATE first (you need git access).

export GOPRIVATE=innovationstream.app/*
go install innovationstream.app/stream-framework/cmd/stream-go@latest

go install drops the binary in $(go env GOPATH)/bin. If your shell reports command not found: stream-go, that directory isn't on your PATH — add it to ~/.zshrc (or ~/.bashrc):

export PATH="$PATH:$(go env GOPATH)/bin"

2. Create a project

stream-go create my-service

The interactive wizard asks for the Go module path, then a preset:

Full — every optional feature plus the example modules (user, auth, notification).
Minimal — REST + GORM only; just the healthcheck module.
Select — tick the features you want.

3. Optional features

healthcheck is always included. Each feature below can be toggled; turning one off deletes its files and strips its env keys.

Feature	What it adds
`grpc`	gRPC server alongside REST
`scheduler`	Cron / scheduled jobs
`mongo`	MongoDB client
`queue`	Background queue worker
`sentry`	Sentry error tracking (optional package)
`ftp`	FTP client (optional package)
`user / auth / notification`	example modules — pick each independently in the wizard checkbox (notification requires `queue`)

4. Non-interactive

Pass --yes with flags to skip the wizard — useful in CI or scripts.

# Everything on
stream-go create my-service --yes --module github.com/me/my-service --preset full

# Minimal
stream-go create my-service --yes --module github.com/me/my-service --preset minimal

# Pick features
stream-go create my-service --yes --module github.com/me/my-service \
  --preset select --grpc --scheduler

5. Running a project

From inside a generated project, stream-go also wraps the dev loop (regenerates Swagger docs first; swag steps are skipped if swag isn't installed):

Command	Does
`stream-go dev`	swag fmt → swag init → `go run .`
`stream-go start`	swag fmt → swag init → `go build -o main .` → `./main` (production-like)
`stream-go update`	reinstall the latest release (`go install …@latest`); the CLI also nudges when a newer version exists

Prefer cloning? You still can: git clone …stream-framework.git my-service then go mod tidy. The CLI just removes the manual module-path rewrite and feature pruning.

Configuration

All configuration lives in two places: .env.dev / .env files and the config/ package.

Config Files

File	Purpose
`config/app.go`	App name, port, environment
`config/database.go`	PostgreSQL DSN, pool settings
`config/redis.go`	Redis host, port, cache prefix/TTL
`config/queue.go`	Queue driver, table names
`config/session.go`	Session driver, TTL, single-device mode
`config/logger.go`	Log mode (file, s3, gcp, graylog)
`config/sentry.go`	Sentry DSN
`config/open_telemetry.go`	OTLP endpoint
`config/file_system.go`	Storage driver (local, s3)
`config/ftp.go`	FTP server credentials and named servers

Key Environment Variables

# App
APP_NAME=stream-framework
APP_PORT=8000
ENV=development          # local | development | production

# Database
DB_HOST=localhost
DB_PORT=5432
DB_NAME=mydb
DB_USER=postgres
DB_PASSWORD=secret

# Redis
REDIS_HOST=localhost
REDIS_PORT=6379
REDIS_CACHE_PREFIX=app:
REDIS_CACHE_DURATION=60      # minutes

# Queue
QUEUE_DRIVER=database        # database | rabbitmq
QUEUE_TABLE=jobs
QUEUE_FAILED_TABLE=failed_jobs

# Auth
AUTH_SINGLE_DEVICE=false

# Observability
SENTRY_DSN=
OTEL_EXPORTER_OTLP_ENDPOINT=http://localhost:4318
LOG_MODE=                    # file | s3 | gcp | graylog

# FTP (default server)
FTP_HOST=
FTP_PORT=21
FTP_USERNAME=
FTP_PASSWORD=
# Named FTP servers (comma-separated)
FTP_SERVERS=                 # e.g. vendor_a,vendor_b
# FTP_VENDOR_A_HOST=
# FTP_VENDOR_A_PORT=21
# FTP_VENDOR_A_USERNAME=
# FTP_VENDOR_A_PASSWORD=

Always update both .env.dev and .env.example when adding new environment variables.

Directory Structure

stream-framework/
├── app/
│   ├── app.go                  # ApplicationServer
│   ├── bootstrap.go            # app.Boot() entry point
│   ├── http_server/            # Fiber server wrapper
│   ├── grpc_server/            # gRPC server wrapper
│   └── scheduler/              # Cron scheduler
├── cmd/
│   ├── gen/                    # Code generator binary
│   ├── queue_worker/           # Queue worker binary
│   └── queue_failed/           # Failed job manager
├── config/                     # One file per config concern
├── database/
│   └── migrations/             # golang-migrate SQL files
├── docs/                       # Auto-generated Swagger docs
├── internal/
│   ├── exception/              # Error mapping helpers
│   ├── jwt/                    # JWT signing/validation
│   ├── hash/                   # Password hashing
│   └── validator/              # Request validation
├── pkg/
│   ├── cache/                  # Redis cache wrapper
│   ├── database/               # GORM + paginator
│   ├── logger/                 # Zap structured logger
│   ├── queue/                  # Queue drivers + worker
│   ├── sentry/                 # Sentry integration
│   ├── session/                # Session store
│   ├── storage/                # Local + S3 file storage
│   └── tracing/                # OpenTelemetry tracer
├── src/
│   ├── middlewares/            # Auth, authz middleware
│   ├── models/                 # GORM models
│   ├── modules/                # Domain modules
│   └── routes/
│       └── http.go             # Route registration
├── main.go
├── .env.dev
└── .env.example

HTTP Layer

Built-in Endpoints

Path	Description
`GET /`	Returns the application name
`GET /health`	Database connection healthcheck
`GET /monitor`	Real-time metrics dashboard
`GET /swagger/*`	Swagger UI (API docs)
`GET /debug/pprof`	Go pprof profiling
`GET /cache/clear`	Flush Redis cache
`GET /`	Framework landing page
`GET /docs`	Framework documentation

Route Registration

func HTTPRoutes(s *http_server.HttpServer) {
    HTTPRootMiddleware(s.Router)
    middleware := middlewares.NewMiddleware()

    api := s.Route().Group("api")

    // Register module routes
    server_module.NewModule().Routes(s.MainRoute(), middleware)
    healthcheck_module.NewModule().Routes(s.MainRoute(), middleware)
    auth_module.NewModule().Routes(api, middleware)
    user_module.NewModule().Routes(api, middleware)
}

Middleware

Global Middleware

Middleware	Purpose
`requestid`	Assigns unique X-Request-ID to every request
`etag`	Adds ETag headers for GET responses
`cors`	Cross-Origin Resource Sharing headers
`logger`	HTTP access log
`recover`	Catches panics and returns 500
`pprof`	Go profiling at `/debug/pprof`

Authentication Middleware

// JwtAuthProtected validates the Bearer token and populates "authUser" in context.
products.Use(mw.JwtAuthProtected())

// Access the authenticated user inside a handler:
authUser := c.Locals("authUser").(*models.User)

Authorization Middleware

route.Use(mw.HasPermissions("products.create"))
route.Use(mw.HasRoles("admin", "manager"))

Module Pattern

Every domain feature follows a strict four-layer pattern:

flowchart LR A[Module] --> B[Controller] B --> C[Service] C --> D[Repository]

Layers

Layer	File	Responsibility
Module	`*_module.go`	Wires dependencies; provides constructor options
Controller	`*_controller.go`	Handles HTTP requests, input validation, response mapping
Service	`*_service.go`	Business logic; orchestrates repository calls
Repository	`*_repository.go`	Database/external I/O; no business logic

Route File (`*_route.go`)

Defines URL paths for the module and binds JWT auth + permission middleware to each handler. Every module must also register its routes in src/routes/http.go.

package product_module

import (
    "github.com/gofiber/fiber/v2"
    "innovationstream.app/stream-framework/src/middlewares"
)

func (m module) Routes(route fiber.Router, middleware *middlewares.Middleware) {
    v1 := route.Group("v1")
    product := v1.Group("products")

    // GET /api/v1/products — ต้อง login + มี permission "product:view"
    product.Get("", middleware.JwtAuthProtected(), middleware.HasPermissions([]string{"product:view"}),
        func(c *fiber.Ctx) error { return m.Controller().GetProducts(c) })

    // POST /api/v1/products — ต้อง login + มี permission "product:create"
    product.Post("", middleware.JwtAuthProtected(), middleware.HasPermissions([]string{"product:create"}),
        func(c *fiber.Ctx) error { return m.Controller().CreateProduct(c) })

    // PUT /api/v1/products/:id
    product.Put(":id", middleware.JwtAuthProtected(), middleware.HasPermissions([]string{"product:update"}),
        func(c *fiber.Ctx) error { return m.Controller().UpdateProduct(c) })

    // DELETE /api/v1/products/:id
    product.Delete(":id", middleware.JwtAuthProtected(), middleware.HasPermissions([]string{"product:destroy"}),
        func(c *fiber.Ctx) error { return m.Controller().DeleteProduct(c) })
}

Controller (`*_controller.go`)

Accepts the HTTP request, parses the body, validates input with validator.Validate, calls the service, and maps errors to HTTP responses using exception.HttpErrorResponseMapping.

package product_module

import (
    "github.com/gofiber/fiber/v2"
    "go.opentelemetry.io/otel/attribute"
    "go.opentelemetry.io/otel/trace"
    "innovationstream.app/stream-framework/internal/exception"
    "innovationstream.app/stream-framework/internal/http_response"
    "innovationstream.app/stream-framework/internal/validator"
    "innovationstream.app/stream-framework/src/modules/product_module/dtos"
)

func (c Controller) CreateProduct(f *fiber.Ctx) error {
    var (
        // เริ่ม tracing span สำหรับ controller นี้
        ctx, span = c.m.tracer.TraceStart(f.Context(), "CreateProductController",
            trace.WithAttributes(attribute.String("controller", "CreateProduct")))
        err error
    )

    // 1. Parse HTTP request body → DTO struct
    dto := new(dtos.CreateProduct)
    if err = f.BodyParser(dto); err != nil {
        return exception.HttpErrorResponseMapping(f, fiber.StatusBadRequest, ProductCreateFailedResponseError, err)
    }

    // 2. Validate DTO ตาม tag ที่กำหนดไว้ใน struct
    if errors := validator.Validate(*dto); errors != nil {
        return exception.HttpErrorResponseMapping(f, fiber.StatusBadRequest,
            exception.InvalidRequestParameterResponseError, exception.ErrInvalidRequestParameter, errors...)
    }

    // 3. เรียก service layer — controller ไม่รู้จัก database
    err = c.productService().CreateProduct(ctx, dto)
    if err != nil {
        return exception.HttpErrorResponseMapping(f, fiber.StatusInternalServerError, ProductCreateFailedResponseError, err)
    }

    // 4. Clear cache ที่เกี่ยวข้อง
    c.m.cacher.Tag("products").Flush(ctx)

    c.m.tracer.TraceEnd(span)
    return http_response.HttpOkResponse(f, "OK", "Product created successfully")
}

Service (`*_service.go`)

Pure business logic — maps DTOs to models, calls the repository, knows nothing about HTTP or response formats. An injectable repository factory makes unit testing straightforward.

package product_module

import (
    "context"

    "go.opentelemetry.io/otel/attribute"
    "go.opentelemetry.io/otel/trace"
    "innovationstream.app/stream-framework/src/models"
    "innovationstream.app/stream-framework/src/modules/product_module/dtos"
    "innovationstream.app/stream-framework/src/modules/product_module/responses"
)

func (s Service) CreateProduct(ctx context.Context, dto *dtos.CreateProduct) error {
    ctx, span := s.tracer.TraceStart(ctx, "CreateProductService",
        trace.WithAttributes(attribute.String("service", "CreateProduct")))

    // Map DTO → model (service รู้จัก business rule, ไม่รู้จัก HTTP)
    product := &models.Product{
        Name:  dto.Name,
        Price: dto.Price,
        Stock: dto.Stock,
    }

    // ส่งต่อให้ repository จัดการ database
    err := s.productRepository().CreateProduct(ctx, product)

    s.tracer.TraceEnd(span)
    return err
}

func (s Service) GetProduct(ctx context.Context, id int) (*responses.GetProductByIDResponse, error) {
    ctx, span := s.tracer.TraceStart(ctx, "GetProductService",
        trace.WithAttributes(attribute.String("service", "GetProduct")))

    product, err := s.productRepository().GetProductByID(ctx, id)

    s.tracer.TraceEnd(span)
    return new(responses.GetProductByIDResponse).Response(product), err
}

Repository (`*_repository.go`)

GORM database access only — uses transactions for writes, wraps errors with Sentry + logger, and converts gorm.ErrRecordNotFound to exception.ErrRecordNotFound so callers can handle it explicitly.

package product_module

import (
    "context"

    "github.com/getsentry/sentry-go"
    "go.opentelemetry.io/otel/attribute"
    "go.opentelemetry.io/otel/trace"
    "gorm.io/gorm"
    "innovationstream.app/stream-framework/internal/exception"
    "innovationstream.app/stream-framework/internal/utils"
    "innovationstream.app/stream-framework/src/models"
)

func (r Repository) CreateProduct(ctx context.Context, product *models.Product) error {
    var (
        _, span = r.tracer.TraceStart(ctx, "CreateProductRepository",
            trace.WithAttributes(attribute.String("repository", "CreateProduct")))
        err error
    )

    utils.Block{
        Try: func() {
            // ใช้ transaction ทุกครั้งที่ write ข้อมูล
            if err = r.db.Transaction(func(tx *gorm.DB) error {
                return tx.Create(product).Error
            }); err != nil {
                utils.Throw(err)
            }
        },
        Catch: func(e utils.Exception) {
            err = e.(error)
            r.logger.Error(err.Error())      // บันทึก log ด้วย Zap
            sentry.CaptureException(err)     // ส่ง error ไป Sentry
            exception.SqlErrorMessage = err.Error()
            err = exception.ErrDbQueryStatement
        },
    }.Do()

    r.tracer.TraceEnd(span)
    return err
}

func (r Repository) GetProductByID(ctx context.Context, id int) (models.Product, error) {
    var (
        _, span  = r.tracer.TraceStart(ctx, "GetProductByIDRepository",
            trace.WithAttributes(attribute.String("repository", "GetProductByID")))
        product models.Product
        err     error
    )

    utils.Block{
        Try: func() {
            if err = r.db.First(&product, id).Error; err != nil {
                utils.Throw(err)
            }
        },
        Catch: func(e utils.Exception) {
            // แยก not-found ออกจาก error อื่น ๆ
            if err == gorm.ErrRecordNotFound {
                err = exception.ErrRecordNotFound
            } else {
                err = e.(error)
                r.logger.Error(err.Error())
                sentry.CaptureException(err)
                exception.SqlErrorMessage = err.Error()
                err = exception.ErrDbQueryStatement
            }
        },
    }.Do()

    r.tracer.TraceEnd(span)
    return product, err
}

Service Options Pattern

func NewService(opts ...ServiceOption) Service
func WithUserRepositoryFactory(factory func() UserRepository) ServiceOption

// In tests:
svc := NewService(WithUserRepositoryFactory(func() UserRepository {
    return &fakeUserRepo{...}
}))

Creating Modules

Using the Code Generator

go build ./cmd/gen

./gen module product                        # Full CRUD module
./gen module product controller,service     # Specific components
./gen model product                         # GORM model only

Generated Files

src/modules/product_module/
├── product_module.go
├── product_controller.go
├── product_service.go
├── product_repository.go
├── product_route.go
├── product_error.go
├── dtos/dtos.go
├── resource/product_resource.go
└── responses/

After Generation

Register the module's routes in src/routes/http.go
Create a database migration in database/migrations/
Create a GORM model in src/models/
Implement service and repository methods

WebSocket Integration

A module can expose a real-time push channel alongside its REST endpoints. The notification module is the reference implementation — it broadcasts events to all connected clients whenever a notification is created or updated.

How It Works

The flow from an HTTP/Queue action to a connected browser is:

flowchart TD A["POST /api/v1/notifications"] --> B["Controller → Service"] B --> C{mode?} C -->|database| D["Repository.Create()"] C -->|queue| E["Queue.Enqueue()"] D --> F["notifyBroadcast()"] E --> G["Worker picks up job"] G --> H["Repository.Create()"] H --> F F --> I["broadcastWS() iterates wsClients"] I --> J["conn1 (WebSocket client)"] I --> K["conn2 (WebSocket client)"]

File Layout

src/modules/notification_module/
├── notification_ws.go      # Client registry + broadcast + WS handler
├── notification_route.go   # Route with WebSocket upgrade
├── notification_service.go # Calls notifyBroadcast() after DB write
└── notification_worker.go  # Queue handler also calls notifyBroadcast()

Client Registry (`*_ws.go`)

A package-level map protected by a sync.RWMutex tracks all open connections. registerWSClient / unregisterWSClient are called inside the handler. broadcastWS serialises a NotificationWSMessage to JSON and writes it to every client, dropping dead connections automatically.

// notification_ws.go
var (
    wsClients   = make(map[*websocket.Conn]bool)
    wsClientsMu sync.RWMutex
)

type NotificationWSMessage struct {
    Type      string      `json:"type"`
    Data      interface{} `json:"data,omitempty"`
    Timestamp time.Time   `json:"timestamp"`
}

func WebSocketNotificationHandler(c *websocket.Conn) {
    registerWSClient(c)
    defer func() {
        unregisterWSClient(c)
        c.Close()
    }()
    for { // keep-alive read loop
        if _, _, err := c.ReadMessage(); err != nil {
            break
        }
    }
}

// Package-internal helper used by Service and Worker
var notifyBroadcast = func(eventType string, data interface{}) {
    broadcastWS(NotificationWSMessage{
        Type:      eventType,
        Data:      data,
        Timestamp: time.Now(),
    })
}

Route Registration (`*_route.go`)

The WebSocket endpoint sits next to the REST routes in the same Routes() method. The upgrade-check middleware must run before websocket.New().

// notification_route.go
notification.Get("ws", func(c *fiber.Ctx) error {
    if websocket.IsWebSocketUpgrade(c) {
        return c.Next()
    }
    return c.SendStatus(fiber.StatusUpgradeRequired)
}, websocket.New(WebSocketNotificationHandler))
// Endpoint: GET /api/v1/notifications/ws

Triggering Broadcasts from the Service

After a successful database write the service calls notifyBroadcast() with an event type and a plain map payload. This keeps the WebSocket concern isolated to *_ws.go while the service layer remains focused on business logic.

// notification_service.go  (direct mode)
err := s.notificationRepository().CreateNotification(ctx, notification)
if err == nil {
    notifyBroadcast("notification.created", map[string]interface{}{
        "title":     notification.Title,
        "recipient": notification.Recipient,
        "message":   notification.Message,
        "timestamp": time.Now().UTC().Unix(),
    })
}

// ReadNotification also broadcasts an update event
notifyBroadcast("notification.updated", map[string]interface{}{"id": id})

Triggering Broadcasts from the Queue Worker

When APP_NOTIFICATION_MODE=queue, the service only enqueues a job. The worker (*_worker.go) is responsible for both persisting the model and broadcasting the event — keeping the same end-to-end behaviour regardless of mode.

// notification_worker.go
func HandleNotificationCreate(ctx context.Context, job queue.Job) error {
    // … build and save model …
    if err := NewRepository().CreateNotification(ctx, notification); err != nil {
        return err
    }
    notifyBroadcast("notification.created", map[string]interface{}{
        "title":     notification.Title,
        "recipient": notification.Recipient,
        "message":   notification.Message,
        "timestamp": time.Now().UTC().Unix(),
    })
    return nil
}

Client-Side Connection Example

Receiving events from the server (one-way broadcast):

const ws = new WebSocket("ws://localhost:3000/api/v1/notifications/ws");

ws.onmessage = (event) => {
    const msg = JSON.parse(event.data);
    // msg.type  → "notification.created" | "notification.updated"
    // msg.data  → payload object
    // msg.timestamp → ISO timestamp
    console.log(msg.type, msg.data);
};

ws.onclose = () => console.log("WebSocket closed");

Sending Messages from Client to Server

By default the server read loop discards all incoming messages. To handle client-to-server messages, extend WebSocketNotificationHandler to parse each frame and dispatch on msg.type:

// notification_ws.go — extended handler
type clientMessage struct {
    Type string          `json:"type"`
    Data json.RawMessage `json:"data,omitempty"`
}

func WebSocketNotificationHandler(c *websocket.Conn) {
    registerWSClient(c)
    defer func() {
        unregisterWSClient(c)
        c.Close()
    }()

    for {
        _, raw, err := c.ReadMessage()
        if err != nil {
            break // client disconnected
        }

        var msg clientMessage
        if err := json.Unmarshal(raw, &msg); err != nil {
            continue // ignore malformed frames
        }

        switch msg.Type {
        case "ping":
            _ = c.WriteJSON(NotificationWSMessage{
                Type:      "pong",
                Timestamp: time.Now(),
            })

        case "mark_read":
            var payload struct {
                ID int `json:"id"`
            }
            if err := json.Unmarshal(msg.Data, &payload); err == nil {
                // call service/repository as needed
                _ = c.WriteJSON(NotificationWSMessage{
                    Type:      "mark_read.ack",
                    Data:      map[string]int{"id": payload.ID},
                    Timestamp: time.Now(),
                })
            }
        }
    }
}

On the client, use ws.send() to push a JSON frame at any time after the connection is open:

// Send a ping to verify the connection is alive
ws.send(JSON.stringify({ type: "ping" }));

// Ask the server to mark notification #42 as read
ws.send(JSON.stringify({ type: "mark_read", data: { id: 42 } }));

// Handle server replies (pong, ack, etc.)
ws.onmessage = (event) => {
    const msg = JSON.parse(event.data);
    switch (msg.type) {
        case "pong":
            console.log("Server is alive");
            break;
        case "mark_read.ack":
            console.log("Marked as read:", msg.data.id);
            break;
        case "notification.created":
            showNotification(msg.data);
            break;
    }
};

The client registry is in-process only. For horizontally scaled deployments, replace the in-memory map with a pub/sub backend (e.g. Redis Pub/Sub) so that broadcasts from any instance reach all connected clients.

Route Registration

src/routes/http.go is the only place to register new module routes.

func HTTPRoutes(s *http_server.HttpServer) {
    api := s.Route().Group("api")
    // Add your module here:
    product_module.NewModule().Routes(api, middleware)
}

Route Group Conventions

Group	Path prefix	Use for
`s.MainRoute()`	`/`	Top-level routes (health, docs, server)
`api`	`/api`	Versioned API endpoints

Prompt Guide

Write prompts that let LLMs generate complete, production-ready modules from business logic descriptions

This guide provides ready-to-use prompt templates for asking an LLM (Claude, GPT-4, etc.) to generate a full Stream Framework module. Copy a template, fill in your business domain details, and the LLM will produce controller, service, repository, DTOs, model, migration, and test files.

Anatomy of a Good Module Prompt

An effective module prompt contains four parts: context (framework rules), domain (what the module does), schema (data shape), and rules (constraints and edge cases).

## [1] Framework Context
You are generating a Go module for Stream Framework.
Rules:
- Module: innovationstream.app/stream-framework
- Architecture: Controller → Service → Repository (never skip a layer)
- Controllers must call exception.HttpErrorResponseMapping for errors
- Services must accept repository via WithXxxRepositoryFactory option
- Use GORM with BaseModel (ID + timestamps); table prefix is tbl_
- Register routes in src/routes/http.go

## [2] Domain
Generate a [MODULE_NAME] module that handles [DESCRIBE WHAT IT DOES].

## [3] Schema
Fields:
- [field_name] [type] — [description, constraints]

## [4] Rules
- [business rule 1]
- [business rule 2]

## [5] Generate
Produce these files:
- src/models/[name].go
- database/migrations/YYYYMMDDHHMMSS_create_tbl_[name].up.sql
- src/modules/[name]_module/dtos/[name]_dto.go
- src/modules/[name]_module/[name]_controller.go
- src/modules/[name]_module/[name]_service.go
- src/modules/[name]_module/[name]_repository.go
- src/modules/[name]_module/[name]_route.go
- src/modules/[name]_module/[name]_module.go
- src/modules/[name]_module/[name]_test.go

Minimal Module Prompt

ตัวอย่าง prompt สั้นสำหรับ module เดี่ยวที่ไม่ซับซ้อน:

You are generating a Go module for Stream Framework (innovationstream.app/stream-framework).
Architecture: Controller → Service → Repository. GORM, tbl_ prefix, BaseModel.
Controllers use exception.HttpErrorResponseMapping. Services accept repositories via options.

Generate a `product` module for managing products in an e-commerce store.

Schema:
- name        string   — required, unique per store
- description string   — optional
- price       float64  — required, must be > 0
- stock_qty   int      — default 0
- store_id    uint     — FK to tbl_stores (belongs to Store)
- is_active   bool     — default true

Rules:
- Only the store owner (JWT auth user) may create/update/delete products
- Listing supports pagination (page, limit) and search by name
- Deleting a product with stock_qty > 0 must return a 422 error

Generate:
src/models/product.go
database/migrations/20260316000000_create_tbl_products.up.sql
src/modules/product_module/dtos/product_dto.go
src/modules/product_module/product_controller.go
src/modules/product_module/product_service.go
src/modules/product_module/product_repository.go
src/modules/product_module/product_route.go
src/modules/product_module/product_module.go
src/modules/product_module/product_test.go

Full CRUD Module with Relations

ตัวอย่าง prompt สำหรับ module ที่มี relation ซับซ้อนและ business rule หลายข้อ:

You are generating a Go module for Stream Framework (innovationstream.app/stream-framework).
Architecture: Controller → Service → Repository. GORM, tbl_ prefix, BaseModel.
Controllers use exception.HttpErrorResponseMapping. Services accept repositories via options.
Use HasPermissions middleware for authorization. Register routes in src/routes/http.go.

Generate an `order` module for a food-delivery platform.

Schema (tbl_orders):
- user_id        uint     — FK to tbl_users (belongs to User, the customer)
- store_id       uint     — FK to tbl_stores (belongs to Store)
- status         string   — enum: pending | confirmed | preparing | delivered | cancelled
- total_price    float64  — calculated from order items
- delivery_addr  string   — required, max 500 chars
- note           string   — optional

Schema (tbl_order_items):
- order_id    uint    — FK to tbl_orders
- product_id  uint    — FK to tbl_products
- quantity    int     — min 1
- unit_price  float64 — snapshot of product price at order time

Relations:
- Order has many OrderItems
- Order belongs to User (customer) and Store

Business Rules:
- Only authenticated users (JWT) can create orders
- status transitions: pending→confirmed, confirmed→preparing, preparing→delivered, any→cancelled
- Cancellation is only allowed within 5 minutes of order creation
- total_price must equal sum(unit_price * quantity) across all items
- Store owner can update status; customer can only cancel
- List endpoint returns paginated orders filtered by caller role (customer sees own orders, store sees store orders)

Queue:
- On status change, dispatch a job `order.status_changed` with payload {order_id, new_status, user_id}

Generate all files including migration and test.

Bug Fix / Code Review Prompt

ใช้ prompt นี้เมื่อต้องการให้ LLM ช่วย review หรือ fix โค้ดที่มีอยู่แล้ว:

You are reviewing a Go module in Stream Framework (innovationstream.app/stream-framework).
Architecture: Controller → Service → Repository.
Identify and fix any issues with:
1. Missing error handling or wrong HTTP status codes
2. Business logic leaking into controller or repository layer
3. Missing or incorrect JWT/permission middleware
4. N+1 query problems in GORM (missing Preload)
5. Race conditions in concurrent operations
6. Missing test coverage for edge cases

Here is the code to review:

--- [PASTE YOUR FILE CONTENTS HERE] ---

After analysis:
1. List each issue found (layer, line range, description)
2. Provide the corrected file(s) in full
3. Explain what changed and why

Tips for Better Results

Always include the framework context

Start every prompt with the framework rules block so the LLM generates idiomatic code that follows the correct layering and conventions.

Describe business rules explicitly

Name every constraint (unique fields, status transitions, ownership checks) — the LLM cannot infer business rules it has not been told.

Specify what to generate

List the exact files you want (model, migration, DTO, controller, service, repository, route, test). Omitting a file means it will not be generated.

Models & GORM

Models live in src/models/. GORM applies a tbl_ prefix to all table names.

Base Models

// BaseModel — ID + timestamps
type BaseModel struct {
    ID        uint           `gorm:"primaryKey"`
    CreatedAt time.Time
    UpdatedAt time.Time
    DeletedAt gorm.DeletedAt `gorm:"index"`
}

// BaseAuditorModel — adds created/updated by user ID
type BaseAuditorModel struct {
    BaseModel
    CreatedBy *uint
    UpdatedBy *uint
}

Creating a Model

type Product struct {
    BaseModel
    Name   string  `gorm:"not null"`
    Price  float64 `gorm:"not null"`
    UserID uint
    User   User    `gorm:"foreignKey:UserID"`
}

func (Product) TableName() string { return "tbl_products" }

Migrations

Schema is managed with golang-migrate using SQL files in database/migrations/.

database/migrations/
├── 000001_create_users_table.up.sql
├── 000001_create_users_table.down.sql
└── ...

-- 000003_create_products_table.up.sql
CREATE TABLE tbl_products (
    id         SERIAL PRIMARY KEY,
    name       VARCHAR(255) NOT NULL,
    price      DECIMAL(10,2) NOT NULL,
    user_id    INTEGER REFERENCES tbl_users(id),
    created_at TIMESTAMP WITH TIME ZONE DEFAULT NOW(),
    updated_at TIMESTAMP WITH TIME ZONE DEFAULT NOW(),
    deleted_at TIMESTAMP WITH TIME ZONE
);

# Apply all pending migrations
migrate -path database/migrations -database "$DATABASE_URL" up

# Roll back last migration
migrate -path database/migrations -database "$DATABASE_URL" down 1

JWT Authentication

func (m module) Routes(route fiber.Router, mw *middlewares.Middleware) {
    products := route.Group("v1/products")
    products.Get("/", c.GetProducts)           // Public

    products.Use(mw.JwtAuthProtected())        // Protected below
    products.Post("/", c.CreateProduct)
}

func (c Controller) CreateProduct(f *fiber.Ctx) error {
    authUser := f.Locals("authUser").(*models.User)
    _ = authUser.ID
}

Session Management

SESSION_DRIVER=redis       # redis | database
SESSION_TTL=86400          # seconds (24h)
AUTH_SINGLE_DEVICE=false   # one active session per user

When AUTH_SINGLE_DEVICE=true, logging in from a new device automatically invalidates all previous sessions for that user.

Authorization

// Permission-based
products.Use(mw.HasPermissions("products.create"))

// Role-based
admin.Use(mw.HasRoles("admin", "manager"))

Always apply JwtAuthProtected() before HasPermissions() or HasRoles().

Queue System

Driver	ENV	Notes
Database	`database`	PostgreSQL with SKIP LOCKED — no extra infra
RabbitMQ	`rabbitmq`	High-throughput workloads

err := queue.AppQueue.Dispatch("notification.create", map[string]any{
    "user_id": user.ID,
    "message": "Welcome!",
})

Job Handlers

// notification_worker.go
package notification_module

import "innovationstream.app/stream-framework/pkg/queue"

func init() {
    queue.RegisterHandler("notification.create", HandleNotificationCreate)
}

func HandleNotificationCreate(ctx context.Context, payload []byte) error {
    var data struct {
        UserID  uint   `json:"user_id"`
        Message string `json:"message"`
    }
    if err := json.Unmarshal(payload, &data); err != nil {
        return err
    }
    // Business logic...
    return nil
}

Return nil on success (job deleted) · Return error (job moved to failed table). Handlers must be idempotent.

Queue Worker

go build ./cmd/queue_worker
./queue_worker --numprocs=4

The worker also starts automatically when QUEUE_DRIVER=database and the main server boots.

Failed Jobs

go run ./cmd/queue_failed --action=list
go run ./cmd/queue_failed --action=requeue --id=123

OpenTelemetry Tracing

func (c Controller) GetProduct(f *fiber.Ctx) error {
    ctx, span := c.m.tracer.TraceStart(
        f.Context(), "GetProductController",
        trace.WithAttributes(attribute.String("product.id", f.Params("id"))),
    )
    defer c.m.tracer.TraceEnd(span)
    // ...
}

OTEL_EXPORTER_OTLP_ENDPOINT=http://localhost:4318
OTEL_SERVICE_NAME=stream-framework

Sentry

SENTRY_DSN=https://xxx@sentry.io/project-id

import sentryException "innovationstream.app/stream-framework/pkg/sentry"
sentryException.CaptureException(err)

Sentry is only active when SENTRY_DSN is set. Safe to leave empty in development.

Logging

log := logger.CurrentLogger()
log.Info("processing job", zap.String("job_id", id))
log.Error("database error", zap.Error(err))

Use logger.CurrentLogger() instead of fmt.Println. Structured logs are easier to query in production.

LOG_MODE	Output
(empty)	Console stdout only
`file`	Local file on disk
`s3`	AWS S3 bucket (rotated daily)
`gcp`	Google Cloud Logging
`graylog`	Graylog (GELF over UDP)

Monitoring

The framework ships a built-in monitoring console served at /monitor. It combines live process & system metrics with a request inspector, an OpenTelemetry trace timeline, and a live log stream — no external APM required for local debugging.

Open http://localhost:8000/monitor in development. All /monitor routes are protected by HTTP Basic Auth (default user stream / password Ps4*123 — set ADMIN_USERNAME / ADMIN_PASSWORD to override).

What You Get

Feature	Description
Live metrics	Process & system CPU %, RAM, open connections and load average, drawn as sparkline charts (polled at 1 Hz).
Request inspector	A table of recent requests — method, path, status, latency, trace ID. Click a row to open a drawer with query params, headers, and request/response bodies (sensitive fields redacted).
Trace timeline	A Jaeger-style waterfall of the OpenTelemetry spans for the selected request, colored by depth, with attributes on hover.
Live logs	A follow-able, substring-searchable stream of the app's structured logs.

Inspection Mode

The request inspector, trace timeline, and log capture are gated behind MONITOR_INSPECT. It defaults to on in development and off in production, so secrets are never held in memory on a live server. Set it explicitly to override:

# Enable request/trace/log capture (default in dev)
MONITOR_INSPECT=true

# Disable capture — only metrics + dashboard remain
MONITOR_INSPECT=false

With inspection on, a capture middleware records each request into an in-memory ring buffer (defaults: 200 requests, 1,000 log lines, 100 trace IDs; bodies truncated to 16 KB). Everything is lost on restart — this is a debugging aid, not a persistent store.

Environment Variables

Variable	Description
`MONITOR_INSPECT`	Enable request/trace/log capture. Default: on outside production.
`ADMIN_USERNAME`	Basic-auth username for `/monitor/*`. Default: `stream`.
`ADMIN_PASSWORD`	Basic-auth password for `/monitor/`. Default: `Ps4123`.
`HTTP_PORT`	Server port — affects the dashboard URL. Default: `8000`.

Routes

GET /monitor              # HTML dashboard
GET /monitor/stats        # process & system metrics (JSON)
GET /monitor/meta         # { inspect, appName }
# Only when MONITOR_INSPECT=true:
GET /monitor/requests     # captured requests (newest first)
GET /monitor/requests/:id # request detail + spans
GET /monitor/logs         # live log stream

The default credentials stream / Ps4*123 are for local development only. Always set strong ADMIN_USERNAME / ADMIN_PASSWORD before deploying, and keep MONITOR_INSPECT off in production.

File System

The pkg/storage package wraps local disk and S3-compatible storage (MinIO / AWS S3) behind a unified interface. Switch drivers by setting FILE_SYSTEM_DISK — no code changes required.

1. Initialization (app/app.go)

storage.FileStorage = storage.NewFileSystem(
	config.Global.FileSystem.Disk,
	storage.WithS3Endpoint(config.Global.FileSystem.S3Config.Endpoint),
	storage.WithS3Bucket(config.Global.FileSystem.S3Config.Bucket),
	storage.WithS3Region(config.Global.FileSystem.S3Config.Region),
	storage.WithS3AccessKeyID(config.Global.FileSystem.S3Config.AccessKeyID),
	storage.WithS3SecretKey(config.Global.FileSystem.S3Config.SecretKey),
	storage.WithS3UseSSL(config.Global.FileSystem.S3Config.UseSSL),
)

2. Environment Variables

Variable	Description
`FILE_SYSTEM_DISK`	Driver: `local` (default) or `s3`
`FILE_SYSTEM_LOCAL_PATH`	Base path for local storage (e.g. `./storage`)
`FILE_SYSTEM_S3_ENDPOINT`	S3 / MinIO endpoint
`FILE_SYSTEM_S3_BUCKET_NAME`	Bucket name
`FILE_SYSTEM_S3_REGION`	Bucket region
`FILE_SYSTEM_S3_ACCESS_KEY_ID`	Access key ID
`FILE_SYSTEM_S3_SECRET_KEY`	Secret key
`FILE_SYSTEM_S3_USE_SSL`	Enable TLS (default: false)

3. Module Wiring

type Repository struct {
	db         *gorm.DB
	tracer     *tracing.MyTracer
	logger     *logger.Logger
	fileSystem *storage.FileSystem
}

func NewRepository() Repository {
	return Repository{
		db:         database.CurrentDatabase(),
		tracer:     tracing.CurrentTracer(),
		logger:     logger.CurrentLogger(),
		fileSystem: storage.CurrentFileStorage(),
	}
}

4. Repository Usage

// Put — อัปโหลดไฟล์ (*multipart.FileHeader หรือ *os.File)
err := r.fileSystem.Put(ctx, "uploads/images", "photo.jpg", fh)

// Get — ดึงไฟล์กลับ
file, err := r.fileSystem.Get(ctx, "uploads/images/photo.jpg")
defer file.Close()

// Move
err = r.fileSystem.Move(ctx, "tmp/photo.jpg", "uploads/photo.jpg")

// Delete
err = r.fileSystem.Delete(ctx, "uploads/images/photo.jpg")

// Disk — สลับ driver ชั่วคราว
err = r.fileSystem.Disk("s3").Put(ctx, "backups", "dump.sql", file)

FTP

The pkg/ftp package wraps github.com/jlaffaye/ftp and follows the same singleton pattern as other infrastructure packages. Each operation opens a fresh connection and closes it automatically — no persistent connection is held.

1. Initialization (app/app.go)

// Default FTP client
ftp.FtpClient = ftp.NewFTPClient(
	config.Global.FTP.Host,
	config.Global.FTP.Port,
	config.Global.FTP.Username,
	config.Global.FTP.Password,
)

// Named FTP clients — loaded automatically from FTP_SERVERS env var
for name, srv := range config.Global.FTP.Servers {
	ftp.Register(name, ftp.NewFTPClient(srv.Host, srv.Port, srv.Username, srv.Password))
}

2. Environment Variables

# Default FTP server
FTP_HOST=ftp.example.com
FTP_PORT=21
FTP_USERNAME=user
FTP_PASSWORD=secret

# Named FTP servers (comma-separated)
FTP_SERVERS=vendor_a,vendor_b
FTP_VENDOR_A_HOST=ftp.vendor-a.com
FTP_VENDOR_A_PORT=21
FTP_VENDOR_A_USERNAME=user_a
FTP_VENDOR_A_PASSWORD=pass_a

3. Module Wiring

Add ftpClient to the Repository struct and inject it via NewRepository().

import ftppkg "innovationstream.app/stream-framework/pkg/ftp"

type Repository struct {
	db        *gorm.DB
	tracer    *tracing.MyTracer
	logger    *logger.Logger
	ftpClient *ftppkg.Client   // เพิ่ม field นี้
}

func NewRepository() Repository {
	return Repository{
		db:        database.CurrentDatabase(),
		tracer:    tracing.CurrentTracer(),
		logger:    logger.CurrentLogger(),
		ftpClient: ftppkg.CurrentFTPClient(),
	}
}

4. Repository Usage

// Upload
file, _ := fh.Open()
defer file.Close()
r.ftpClient.Upload("remote/path", fh.Filename, file)

// Download — ต้อง Close() ทุกครั้ง
reader, err := r.ftpClient.Download("remote/path/file.txt")
if err != nil { ... }
defer reader.Close()

// Move
r.ftpClient.Move("old/path/file.txt", "new/path/file.txt")

// Named FTP Client — ใช้ ftppkg.Use("name") แทน r.ftpClient
ftppkg.Use("vendor_a").Upload("remote/path", fh.Filename, file)

reader, err = ftppkg.Use("vendor_a").Download("remote/path/file.txt")
if err != nil { ... }
defer reader.Close()

ftppkg.Use("vendor_a").Move("old/path/file.txt", "new/path/file.txt")

5. Named FTP Client

Call ftppkg.Use("name") directly without injecting into the struct.

ftppkg.Use("vendor_a").Upload("remote/path", fh.Filename, file)

Download returns an io.ReadCloser that wraps the FTP connection. Always call defer reader.Close() to avoid connection leaks.

Testing Pattern

Tests inject fake repositories via the WithXxxRepositoryFactory option pattern, avoiding real database connections.

svc := NewService(
    WithUserRepositoryFactory(func() UserRepository {
        return &fakeUserRepo{...}
    }),
)
result, err := svc.GetUserByID(ctx, 1)

Test cases follow the [MODULE-T-NNN] naming convention and are catalogued in TEST_CASES.md.

Writing Tests

Unit Tests

ใช้ fake repository ผ่าน WithXxxRepositoryFactory และ inject ผ่าน constructor option ไม่ต้องใช้ database จริง

go test ./...
go test ./src/modules/user_module/...
go test -v -run TestGetUserByID ./...

// [USR-T-001] GetUsers returns all users
func TestGetUsers(t *testing.T) {
    repo := &fakeUserRepo{
        users: []models.User{
            {BaseModel: models.BaseModel{ID: 1}, Name: "Alice"},
        },
    }
    svc := NewService(WithUserRepositoryFactory(func() UserRepository { return repo }))
    result, err := svc.GetUsers(context.Background())
    assert.NoError(t, err)
    assert.Len(t, result, 1)
}

Contract Testing (Pact)

Contract testing ตรวจสอบว่า Provider ตอบสนองตาม contract ที่ Consumer กำหนดไว้ ป้องกัน breaking change ระหว่าง service โดยไม่ต้อง integration test แบบ end-to-end

Pact tests อยู่ใน src/pact/ และต้องใช้ build tag pact ก่อนรัน ต้องติดตั้ง Pact FFI native library ก่อน 1 ครั้ง

ติดตั้ง Pact FFI (ครั้งแรก)

go install github.com/pact-foundation/pact-go/v2@latest
pact-go -l DEBUG install          # Linux / macOS → /usr/local/lib
# Windows: pact-go -l DEBUG install --libDir C:\pact\lib

รัน Pact Tests

# 1. Consumer test — สร้าง pact contract ไว้ใน pacts/
go test ./src/pact/... -tags pact -run TestUserConsumer

# 2. Provider verification — ตรวจสอบ provider ว่าตรง contract
go test ./src/pact/... -tags pact -run TestUserProviderPact -v

Consumer Test

Consumer กำหนด contract โดยระบุ request ที่จะส่ง และ response shape ที่คาดหวัง Pact จะสร้าง mock server และบันทึก interaction เป็น JSON ไว้ใน pacts/

//go:build pact

// [USR-C-001] GET /api/v1/users/:id คืน 200 และ user resource เมื่อ user มีอยู่
func TestUserConsumerPact_GetUserByID(t *testing.T) {
    p, _ := consumer.NewV2Pact(consumer.MockHTTPProviderConfig{
        Consumer: "UserDashboard",
        Provider: "UserService",
        PactDir:  "../../pacts",
    })

    p.AddInteraction().
        Given("user with id 1 exists").
        UponReceiving("a request to get user with id 1").
        WithRequest("GET", "/api/v1/users/1").
        WillRespondWith(http.StatusOK, func(b *consumer.V2ResponseBuilder) {
            b.Header("Content-Type", matchers.S("application/json"))
            b.JSONBody(map[string]interface{}{
                "code":    matchers.S("OK"),
                "message": matchers.S("Success"),
                "data": map[string]interface{}{
                    "id":        matchers.Integer(1),
                    "email":     matchers.S("john@example.com"),
                    "firstName": matchers.S("John"),
                    "lastName":  matchers.S("Doe"),
                },
            })
        }).
        ExecuteTest(t, func(config consumer.MockServerConfig) error {
            resp, err := http.Get(fmt.Sprintf("http://localhost:%d/api/v1/users/1", config.Port))
            // ...verify status code...
            return err
        })
}

Provider Verification

Provider อ่าน pact file ที่ Consumer สร้างไว้ แล้ว replay ทุก interaction กับ Fiber server จริง (backed by in-memory store) โดยใช้ StateHandlers เพื่อเตรียมข้อมูลก่อนแต่ละ interaction

//go:build pact

// [USR-P-001] ตรวจสอบว่า UserService ตอบสนองตาม contract ที่ UserDashboard กำหนด
func TestUserProviderPact(t *testing.T) {
    store := newUserTestStore()        // in-memory repository
    app   := newPactFiberApp(store)    // Fiber + real UserService
    port, shutdown := startPactServer(t, app)
    defer shutdown()

    verifier := provider.NewVerifier()
    verifier.VerifyProvider(t, provider.VerifyRequest{
        ProviderBaseURL: fmt.Sprintf("http://127.0.0.1:%d", port),
        PactFiles: []string{"../../pacts/UserDashboard-UserService.json"},
        StateHandlers: pactmodels.StateHandlers{
            "user with id 1 exists": func(setup bool, _ pactmodels.ProviderState) (pactmodels.ProviderStateResponse, error) {
                if setup {
                    store.Set(1, testUser1)
                } else {
                    store.Clear()
                }
                return pactmodels.ProviderStateResponse{}, nil
            },
        },
    })
}

โครงสร้างไฟล์

src/pact/
├── user_consumer_pact_test.go   # Consumer: กำหนด contract + สร้าง pact JSON
└── user_provider_pact_test.go   # Provider: verify ตาม contract

pacts/
└── UserDashboard-UserService.json  # Generated — อย่า commit โดยตรง

ต้องรัน Consumer test ก่อนเสมอ เพื่อสร้าง pacts/*.json ก่อนที่ Provider verification จะอ่านได้

E2E Testing

End-to-end tests exercise the real HTTP API against a live server backed by a throwaway PostgreSQL + Redis stack. They run Postman collections with Newman and assert the contract every endpoint actually serves — catching bugs that unit tests, which mock the database, never reach.

Run the whole pipeline with one command. It is self-contained — it boots its own services, migrates, seeds, runs the collections, and tears everything down:

make e2e

Requirements

The harness needs docker, go, and npx (Node.js) on your PATH. Newman is fetched on demand via npx --yes, so nothing is installed globally.

What `make e2e` Does

make e2e runs scripts/e2e.sh, which executes these steps and cleans up on exit (even on failure):

Start an isolated PostgreSQL + Redis via docker-compose.e2e.yml (ports 55432 / 56379 to avoid clashing with your dev stack).
Apply every migration in database/migrations/*.up.sql.
Boot the server with go run . on port 18800 and poll /health until ready.
Seed roles, permissions, and an admin user from database/seeds/e2e_seed.sql.
Run each Postman collection with Newman, in order, chaining the auth token between them.
Run a drift check (postmanlint drift) to confirm every live route is covered by exactly one collection.
Tear down the server and Docker volumes.

Test Layout

Collections live in postman/ — one file per module. They run in order because the auth collection logs in and exports the access token (to /tmp/e2e_env.json) that the later collections reuse:

postman/
├── auth_module.postman_collection.json          # logs in, exports accessToken
├── user_module.postman_collection.json
├── notification_module.postman_collection.json
├── healthcheck_module.postman_collection.json
└── stream-local.postman_environment.json         # baseUrl, tokens, admin creds

Collections must be rerunnable from a clean database. Don't depend on data left by a previous run — pass IDs between requests within the same run using pm.environment.set(...).

Adding or Updating Tests

Every new or changed endpoint ships its Postman entry in the same change set (conventions rule 10). The full workflow — required fields, the two-layer example rule, and lint rules — lives in docs/agent/recipes/update-postman-e2e.md. In short:

Add the request to postman/<module>_module.postman_collection.json (Collection v2.1). Scaffolding a module with go run ./cmd/gen module <name> emits a lint-valid skeleton.
Give it a non-empty description, and a description on every query/form param.
Save at least two examples: one pure-JSON response and one (annotated) example whose comment lines carry // after the last quote.
Run make e2e locally; make verify runs the offline postman-lint structural checks.

make e2e needs Docker running. The pipeline stops on the first failing request (--bail), and the drift check fails the build if a live route has no matching collection entry.

CLI Reference

Development Server

go run .          # Run (loads .env.dev)
air               # Live reload
go build -o main  # Build binary

Testing

go test ./...
go test ./src/modules/user_module/...

Swagger Docs

swag init    # Generate from annotations
swag fmt     # Format comments

Code Generator

go build ./cmd/gen
./gen module <name>
./gen module <name> controller,service
./gen model <name>

Queue Worker

go build ./cmd/queue_worker
./queue_worker --numprocs=4

Failed Job Manager

go run ./cmd/queue_failed --action=list
go run ./cmd/queue_failed --action=requeue --id=123

Stream Framework

Tech Stack

Architecture Overview

Development Life-cycle

ขั้นตอนการพัฒนา

1. เปิด Feature Branch

2. Scaffold Module (ถ้าเป็น Feature ใหม่)

3. ออกแบบ Model & Migration

4. Implement Repository

5. Implement Service

6. Implement Controller

7. ลงทะเบียน Routes

8. เขียน Tests

9. Regenerate Swagger & ทดสอบ Local

ภาพรวมขั้นตอน

Requirements

Runtime

Development Tools

Installation

1. Clone the repository

2. Install Go dependencies

3. Copy environment file

4. Run database migrations

5. Start the server

stream-go CLI

1. Install

2. Create a project

3. Optional features

4. Non-interactive

5. Running a project

Configuration

Config Files

Key Environment Variables

Directory Structure

Boot Sequence

HTTP Layer

Built-in Endpoints

Route Registration

Middleware

Global Middleware

Authentication Middleware

Authorization Middleware

Module Pattern

Layers

Route File (*_route.go)

Controller (*_controller.go)

Service (*_service.go)

Repository (*_repository.go)

Service Options Pattern

Creating Modules

Using the Code Generator

Generated Files

After Generation

WebSocket Integration

How It Works

File Layout

Client Registry (*_ws.go)

Route Registration (*_route.go)

Triggering Broadcasts from the Service

Triggering Broadcasts from the Queue Worker

Client-Side Connection Example

Sending Messages from Client to Server

Route Registration

Route Group Conventions

Prompt Guide

Anatomy of a Good Module Prompt

Minimal Module Prompt

Full CRUD Module with Relations

Bug Fix / Code Review Prompt

Tips for Better Results

Models & GORM

Base Models

Creating a Model

Migrations

JWT Authentication

Session Management

Authorization

Queue System

Job Handlers

Queue Worker

Route File (`*_route.go`)

Controller (`*_controller.go`)

Service (`*_service.go`)

Repository (`*_repository.go`)

Client Registry (`*_ws.go`)

Route Registration (`*_route.go`)

What `make e2e` Does