feat: add domain and repository packages (#1)

.dockerignore

@@ -0,0 +1,3 @@
+/.git/
+/.github/
+/psqlqueue

.github/workflows/test.yml

@@ -0,0 +1,56 @@
+name: test
+
+on:
+  workflow_call:
+  push:
+    branches:
+      - "*"
+
+env:
+  PSQLQUEUE_TEST_DATABASE_URL: "postgres://psqlqueue:[email protected]:5432/psqlqueue-test?sslmode=disable"
+  PSQLQUEUE_TESTING: "true"
+
+jobs:
+  test:
+    runs-on: ubuntu-latest
+    services:
+      postgres:
+        image: postgres:15-alpine
+        env:
+          POSTGRES_DB: psqlqueue-test
+          POSTGRES_USER: psqlqueue
+          POSTGRES_PASSWORD: psqlqueue
+        ports:
+          - 5432:5432
+        options: >-
+          --health-cmd pg_isready
+          --health-interval 10s
+          --health-timeout 5s
+          --health-retries 5
+    steps:
+      - name: Set up Go 1.21
+        uses: actions/setup-go@v5
+        with:
+          go-version: "1.21"
+        id: go
+      - name: Check out code into the Go module directory
+        uses: actions/checkout@v4
+      - name: Set cache
+        uses: actions/cache@v3
+        with:
+          path: |
+            ~/.cache/go-build
+            ~/go/pkg/mod
+          key: ${{ runner.os }}-go-${{ hashFiles('**/go.sum') }}
+      - name: Get dependencies
+        run: go mod download
+      - name: Create default dotenv for testing
+        run: cp env.sample .env
+      - name: golangci-lint
+        uses: golangci/golangci-lint-action@v3
+        with:
+          version: latest
+      - name: Run Database Migration
+        run: make run-migration
+      - name: Run Tests
+        run: make test

.gitignore

@@ -19,3 +19,6 @@
 
 # Go workspace file
 go.work
+
+/.env
+/psqlqueue

.golangci.yml

@@ -0,0 +1,9 @@
+run:
+  timeout: 5m
+linters:
+  enable:
+    - gosec
+    - goimports
+linters-settings:
+  goimports:
+    local-prefixes: github.com/allisson/psqlqueue

Dockerfile

@@ -0,0 +1,24 @@
+#### development stage
+FROM golang:1.21 AS build-env
+
+# set envvar
+ENV CGO_ENABLED=0
+ENV GOOS=linux
+
+# set workdir
+WORKDIR /code
+
+# get project dependencies
+COPY go.mod /code/
+RUN go mod download
+
+# copy files
+COPY . /code
+
+# generate binary
+RUN go build -ldflags="-s -w" -o ./psqlqueue ./cmd/psqlqueue
+
+#### final stage
+FROM gcr.io/distroless/base:nonroot
+COPY --from=build-env /code/psqlqueue /
+ENTRYPOINT ["/psqlqueue"]

Makefile

@@ -0,0 +1,49 @@
+.PHONY: lint
+lint:
+	golangci-lint run -v --fix
+
+.PHONY: test
+test:
+	go test -covermode=count -coverprofile=count.out -v ./...
+
+.PHONY: build
+build:
+	go build -ldflags="-s -w" -o ./psqlqueue ./cmd/psqlqueue
+
+.PHONY: build-image
+build-image:
+	docker build --rm -t psqlqueue .
+
+.PHONY: run-db
+run-db:
+	docker run --name postgres-psqlqueue \
+		--restart unless-stopped \
+		-e POSTGRES_USER=psqlqueue \
+		-e POSTGRES_PASSWORD=psqlqueue \
+		-e POSTGRES_DB=psqlqueue \
+		-p 5432:5432 \
+		-d postgres:15-alpine
+
+.PHONY: rm-db
+rm-db:
+	docker kill $$(docker ps -aqf name=postgres-psqlqueue)
+	docker container rm $$(docker ps -aqf name=postgres-psqlqueue)
+
+.PHONY: run-test-db
+run-test-db:
+	docker run --name postgres-psqlqueue-test \
+		--restart unless-stopped \
+		-e POSTGRES_USER=psqlqueue \
+		-e POSTGRES_PASSWORD=psqlqueue \
+		-e POSTGRES_DB=psqlqueue-test \
+		-p 5432:5432 \
+		-d postgres:15-alpine
+
+.PHONY: rm-test-db
+rm-test-db:
+	docker kill $$(docker ps -aqf name=postgres-psqlqueue-test)
+	docker container rm $$(docker ps -aqf name=postgres-psqlqueue-test)
+
+.PHONY: run-migration
+run-migration:
+	go run cmd/psqlqueue/main.go migrate

cmd/psqlqueue/main.go

@@ -0,0 +1,38 @@
+package main
+
+import (
+	"log/slog"
+	"os"
+
+	"github.com/urfave/cli/v2"
+
+	"github.com/allisson/psqlqueue/db/migrations"
+	"github.com/allisson/psqlqueue/domain"
+)
+
+func main() {
+	cfg := domain.NewConfig()
+	logger := domain.NewLogger(cfg.LogLevel)
+	slog.SetDefault(logger)
+
+	app := &cli.App{
+		Commands: []*cli.Command{
+			{
+				Name:    "migrate",
+				Aliases: []string{"m"},
+				Usage:   "run database migrate",
+				Action: func(c *cli.Context) error {
+					if cfg.Testing {
+						return migrations.Migrate(cfg.TestDatabaseURL)
+					}
+					return migrations.Migrate(cfg.DatabaseURL)
+				},
+			},
+		},
+	}
+
+	if err := app.Run(os.Args); err != nil {
+		slog.Error("cli app failed", "error", err)
+		os.Exit(1)
+	}
+}

db/migrations/000001_create_tables.down.sql

@@ -0,0 +1,2 @@
+DROP TABLE IF EXISTS queues;
+DROP TABLE IF EXISTS messages;

db/migrations/000001_create_tables.up.sql

@@ -0,0 +1,27 @@
+CREATE TABLE IF NOT EXISTS queues(
+    id VARCHAR PRIMARY KEY NOT NULL,
+    ack_deadline_seconds INT NOT NULL,
+    message_retention_seconds INT NOT NULL,
+    delivery_delay_seconds INT NOT NULL,
+    created_at TIMESTAMPTZ NOT NULL,
+    updated_at TIMESTAMPTZ NOT NULL
+);
+
+CREATE TABLE IF NOT EXISTS messages(
+    id VARCHAR PRIMARY KEY NOT NULL,
+    queue_id VARCHAR NOT NULL,
+    body VARCHAR NOT NULL,
+    label VARCHAR,
+    attributes JSONB,
+    delivery_attempts INT NOT NULL,
+    expired_at TIMESTAMPTZ NOT NULL,
+    scheduled_at TIMESTAMPTZ NOT NULL,
+    created_at TIMESTAMPTZ NOT NULL,
+    updated_at TIMESTAMPTZ NOT NULL,
+    FOREIGN KEY (queue_id) REFERENCES queues (id) ON DELETE CASCADE
+);
+CREATE INDEX IF NOT EXISTS messages_queue_id_idx ON messages (queue_id);
+CREATE INDEX IF NOT EXISTS messages_label_idx ON messages (label);
+CREATE INDEX IF NOT EXISTS messages_expired_at_idx ON messages USING BRIN (expired_at);
+CREATE INDEX IF NOT EXISTS messages_scheduled_at_idx ON messages USING BRIN (scheduled_at);
+CREATE INDEX IF NOT EXISTS messages_expired_at_scheduled_at_idx ON messages USING BRIN (expired_at, scheduled_at);

db/migrations/migration.go

@@ -0,0 +1,37 @@
+package migrations
+
+import (
+	"embed"
+	"log/slog"
+	"strings"
+
+	"github.com/golang-migrate/migrate/v4"
+	_ "github.com/golang-migrate/migrate/v4/database/pgx"
+	"github.com/golang-migrate/migrate/v4/source/iofs"
+)
+
+//go:embed *.sql
+var fs embed.FS
+
+func Migrate(databaseURL string) error {
+	slog.Info("migration process started")
+	defer slog.Info("migration process finished")
+
+	parsedDatabaseURL := strings.ReplaceAll(databaseURL, "postgresql://", "pgx://")
+	parsedDatabaseURL = strings.ReplaceAll(parsedDatabaseURL, "postgres://", "pgx://")
+
+	driver, err := iofs.New(fs, ".")
+	if err != nil {
+		return err
+	}
+
+	m, err := migrate.NewWithSourceInstance("iofs", driver, parsedDatabaseURL)
+	if err != nil {
+		return err
+	}
+
+	if err := m.Up(); err != nil && err != migrate.ErrNoChange {
+		return err
+	}
+	return nil
+}

domain/config.go

@@ -0,0 +1,77 @@
+package domain
+
+import (
+	"log/slog"
+	"os"
+	"path"
+
+	"github.com/allisson/go-env"
+	"github.com/joho/godotenv"
+)
+
+func searchup(dir string, filename string) string {
+	if dir == "/" || dir == "" {
+		return ""
+	}
+
+	if _, err := os.Stat(path.Join(dir, filename)); err == nil {
+		return path.Join(dir, filename)
+	}
+
+	return searchup(path.Dir(dir), filename)
+}
+
+func findDotEnv() string {
+	directory, err := os.Getwd()
+	if err != nil {
+		return ""
+	}
+
+	filename := ".env"
+	return searchup(directory, filename)
+}
+
+func loadDotEnv() bool {
+	dotenv := findDotEnv()
+	if dotenv != "" {
+		slog.Info("Found .env", "file", dotenv)
+		if err := godotenv.Load(dotenv); err != nil {
+			slog.Warn("Can't load .env", "file", dotenv, "error", err)
+			return false
+		}
+		return true
+	}
+	return false
+}
+
+// Config holds all application configuration data.
+type Config struct {
+	Testing                        bool
+	LogLevel                       string
+	ServerHost                     string
+	ServerPort                     uint
+	ServerReadHeaderTimeoutSeconds uint
+	DatabaseURL                    string
+	TestDatabaseURL                string
+	DatabaseMinConns               uint
+	DatabaseMaxConns               uint
+	QueueMaxNumberOfMessages       uint
+}
+
+// NewConfig returns a Config with values loaded from environment variables.
+func NewConfig() *Config {
+	loadDotEnv()
+
+	return &Config{
+		Testing:                        env.GetBool("PSQLQUEUE_TESTING", false),
+		LogLevel:                       env.GetString("PSQLQUEUE_LOG_LEVEL", "info"),
+		ServerHost:                     env.GetString("PSQLQUEUE_SERVER_HOST", "0.0.0.0"),
+		ServerPort:                     env.GetUint("PSQLQUEUE_SERVER_PORT", 8000),
+		ServerReadHeaderTimeoutSeconds: env.GetUint("PSQLQUEUE_SERVER_READ_HEADER_TIMEOUT_SECONDS", 60),
+		DatabaseURL:                    env.GetString("PSQLQUEUE_DATABASE_URL", ""),
+		TestDatabaseURL:                env.GetString("PSQLQUEUE_TEST_DATABASE_URL", ""),
+		DatabaseMinConns:               env.GetUint("PSQLQUEUE_DATABASE_MIN_CONNS", 0),
+		DatabaseMaxConns:               env.GetUint("PSQLQUEUE_DATABASE_MAX_CONNS", 2),
+		QueueMaxNumberOfMessages:       env.GetUint("PSQLQUEUE_QUEUE_MAX_NUMBER_OF_MESSAGES", 10),
+	}
+}

domain/config_test.go

@@ -0,0 +1,11 @@
+package domain
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+)
+
+func TestLoadDotEnv(t *testing.T) {
+	assert.True(t, loadDotEnv())
+}

domain/error.go

@@ -0,0 +1,14 @@
+package domain
+
+import "errors"
+
+var (
+	// ErrQueueAlreadyExists is returned when the queue already exists.
+	ErrQueueAlreadyExists = errors.New("queue already exists")
+	// ErrQueueNotFound is returned when the queue is not found.
+	ErrQueueNotFound = errors.New("queue not found")
+	// ErrMessageAlreadyExists is returned when the message already exists.
+	ErrMessageAlreadyExists = errors.New("message already exists")
+	// ErrMessageNotFound is returned when the message is not found.
+	ErrMessageNotFound = errors.New("message not found")
+)

domain/logger.go

@@ -0,0 +1,25 @@
+package domain
+
+import (
+	"log/slog"
+	"os"
+	"strings"
+)
+
+// NewLogger returns a configured JSON logger.
+func NewLogger(logLevel string) *slog.Logger {
+	var level slog.Level
+	switch strings.ToLower(logLevel) {
+	case "info":
+		level = slog.LevelInfo
+	case "debug":
+		level = slog.LevelDebug
+	case "warn":
+		level = slog.LevelWarn
+	case "error":
+		level = slog.LevelError
+	}
+
+	h := slog.NewJSONHandler(os.Stdout, &slog.HandlerOptions{Level: level})
+	return slog.New(h)
+}