aboutsummaryrefslogtreecommitdiffstats
diff options
context:
space:
mode:
Diffstat
-rw-r--r--cmd/mirumd/config.go49+49 −0
-rw-r--r--cmd/mirumd/main.go395+67 −328
-rw-r--r--cmd/mirumd/server.go76+76 −0
-rw-r--r--cmd/mirumd/server_grpc.go222+222 −0
-rw-r--r--cmd/mirumd/server_web.go60+60 −0
-rw-r--r--cmd/mirumw/main.go2+1 −1
-rw-r--r--go.mod21+19 −2
-rw-r--r--go.sum84+77 −7
-rw-r--r--internal/database/database.go68+68 −0
-rw-r--r--internal/supervisor/supervisor.go8+4 −4
-rw-r--r--internal/supervisor/systemd.go11+9 −2
-rw-r--r--pkg/mirumd.service4+2 −2
-rw-r--r--pkg/mirumd.yaml3+2 −1
13 files changed, 656 insertions, 347 deletions
diff --git a/cmd/mirumd/config.go b/cmd/mirumd/config.go
new file mode 100644
--- /dev/null
+++ b/cmd/mirumd/config.go
@@ -0,0 +1,49 @@
+// Copyright (c) 2026 Nikolay Govorov
+// SPDX-License-Identifier: AGPL-3.0-or-later
+
+package main
+
+import (
+ "fmt"
+ "os"
+
+ "gopkg.in/yaml.v3"
+)
+
+type config struct {
+ WwwAddr string `yaml:"www_addr"`
+ GrpcAddr string `yaml:"grpc_addr"`
+ DatabaseUri string `yaml:"database_uri"`
+ WorkerSecret string `yaml:"secret"`
+
+ GitHubToken string `yaml:"token"`
+ WebhookSecret string `yaml:"webhook_secret"`
+}
+
+func getConfig(filename string) (*config, error) {
+ cfg := &config{
+ GrpcAddr: ":2026",
+ WwwAddr: ":3000",
+ }
+
+ data, err := os.ReadFile(filename)
+ if err != nil {
+ return nil, err
+ }
+
+ if err := yaml.Unmarshal(data, &cfg); err != nil {
+ return nil, err
+ }
+
+ if cfg.DatabaseUri == "" {
+ return nil, fmt.Errorf("error: database_uri is required")
+ }
+ if cfg.WorkerSecret == "" {
+ return nil, fmt.Errorf("error: worker_secret is required")
+ }
+ if cfg.WebhookSecret == "" {
+ return nil, fmt.Errorf("error: webhook_secret is required")
+ }
+
+ return cfg, nil
+}
diff --git a/cmd/mirumd/main.go b/cmd/mirumd/main.go
index b53f567..86fe007 100644
--- a/cmd/mirumd/main.go
+++ b/cmd/mirumd/main.go
@@ -5,372 +5,110 @@ package main
import (
"context"
- "errors"
"flag"
"fmt"
- "io"
"log/slog"
"net"
"net/http"
"os"
- "sync"
"time"
+ "dimidiumlabs/mirum/internal/database"
"dimidiumlabs/mirum/internal/forges"
- "dimidiumlabs/mirum/internal/protocol"
"dimidiumlabs/mirum/internal/protocol/pb"
"dimidiumlabs/mirum/internal/supervisor"
"github.com/coreos/go-systemd/v22/activation"
- "google.golang.org/grpc"
- "google.golang.org/grpc/codes"
- "google.golang.org/grpc/peer"
- "google.golang.org/grpc/stats"
- "google.golang.org/grpc/status"
- "google.golang.org/protobuf/types/known/timestamppb"
- "gopkg.in/yaml.v3"
)
-type config struct {
- GrpcAddr string `yaml:"grpc_addr"`
- WwwAddr string `yaml:"www_addr"`
- Secret string `yaml:"secret"`
- Token string `yaml:"token"`
-}
-
-var cfg = config{
- GrpcAddr: ":2026",
- WwwAddr: ":3000",
-}
-
-var configFile = flag.String("config", "", "path to config file")
-
-type taskMeta struct {
- forge forges.Forge
- event *forges.PushEvent
-}
-
-var taskCounter int64
-
-func nextTaskID() string {
- taskCounter++
- return fmt.Sprintf("task-%d", taskCounter)
-}
-
func main() {
+ configFile := flag.String("config", "", "path to config file")
flag.Parse()
- if *configFile != "" {
- data, err := os.ReadFile(*configFile)
- if err != nil {
- fmt.Fprintln(os.Stderr, err)
- os.Exit(1)
- }
- if err := yaml.Unmarshal(data, &cfg); err != nil {
- fmt.Fprintln(os.Stderr, err)
- os.Exit(1)
- }
+ if *configFile == "" {
+ slog.Error("-config required")
+ os.Exit(1)
}
- if cfg.Token == "" {
- fmt.Fprintln(os.Stderr, "error: token is required")
+ cfg, err := getConfig(*configFile)
+ if err != nil {
+ slog.Error("config parsing failed", "err", err)
os.Exit(1)
}
- forge := &forges.GitHub{Secret: cfg.Secret, Token: cfg.Token}
+ slog.Info("config loaded", "configfile", *configFile)
- srv := &mirumServer{
- secret: []byte(cfg.Secret),
- queue: make(chan *pb.Task, 100),
+ db, err := database.Open(context.Background(), cfg.DatabaseUri)
+ if err != nil {
+ slog.Error("couldn't open database: %w", "err", err)
+ os.Exit(1)
}
+ defer db.Close()
- mux := http.NewServeMux()
- mux.HandleFunc("GET /", func(w http.ResponseWriter, r *http.Request) {
- w.Header().Set("Content-Type", "text/html; charset=utf-8")
- fmt.Fprint(w, `<!DOCTYPE html><html><head><meta charset="utf-8"><title>Mirum</title></head><body><h1>Mirum</h1><p>CI server is running.</p></body></html>`)
- })
- mux.HandleFunc("POST /webhook", func(w http.ResponseWriter, r *http.Request) {
- body, err := io.ReadAll(r.Body)
- if err != nil {
- http.Error(w, "read body", http.StatusBadRequest)
- return
- }
-
- ev, err := forge.Webhook(r, body)
- if errors.Is(err, forges.ErrInvalidSignature) {
- http.Error(w, "invalid signature", http.StatusUnauthorized)
- return
- }
- if err != nil {
- http.Error(w, err.Error(), http.StatusBadRequest)
- return
- }
- if ev == nil {
- w.WriteHeader(http.StatusNoContent)
- return
- }
-
- id := nextTaskID()
- slog.Info("push", "repo", ev.Owner+"/"+ev.Repo, "branch", ev.Branch, "sha", ev.SHA[:8], "task", id)
-
- srv.tasks.Store(id, taskMeta{forge: forge, event: ev})
- _ = forge.SetStatus(context.Background(), ev, forges.StatusPending, "Queued")
-
- srv.queue <- &pb.Task{
- Id: id,
- CloneUrl: forge.AuthURL(ev.CloneURL),
- Branch: ev.Branch,
- Sha: ev.SHA,
- RepoFullName: ev.Owner + "/" + ev.Repo,
- }
-
- w.WriteHeader(http.StatusAccepted)
- })
-
- grpcLn, httpLn, err := listeners()
- if err != nil {
- fmt.Fprintln(os.Stderr, err)
+ if err := db.Migrate(context.Background()); err != nil {
+ slog.Error("migration failed: %w", "err", err)
os.Exit(1)
}
- grpcSrv := grpc.NewServer(
- grpc.UnaryInterceptor(srv.unaryInterceptor),
- grpc.StreamInterceptor(srv.streamInterceptor),
- grpc.StatsHandler(&connTracker{server: srv}),
- )
- pb.RegisterMirumServer(grpcSrv, srv)
- httpSrv := &http.Server{Handler: mux}
+ slog.Info("database ready")
- slog.Info("listening", "grpc", grpcLn.Addr(), "http", httpLn.Addr())
+ srv := &server{
+ cfg: cfg,
+ db: db,
+ forge: &forges.GitHub{Secret: cfg.WebhookSecret, Token: cfg.GitHubToken},
+ queue: make(chan *pb.Task, 100),
+ }
sup := supervisor.Detect()
ctx := sup.WaitForStop(context.Background())
- go func() {
- <-ctx.Done()
- slog.Info("shutting down")
- sup.Stopping()
+ wwwSrv := NewWwwServer(ctx, srv)
+ grpcSrv := NewGrpcServer(ctx, srv, []byte(cfg.WorkerSecret))
- grpcSrv.GracefulStop()
-
- shutdownCtx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
- defer cancel()
- httpSrv.Shutdown(shutdownCtx)
- }()
-
- go grpcSrv.Serve(grpcLn)
-
- sup.Ready()
- go sup.StartWatchdog()
-
- if err := httpSrv.Serve(httpLn); err != http.ErrServerClosed {
+ grpcLn, webLn, err := listeners(cfg)
+ if err != nil {
fmt.Fprintln(os.Stderr, err)
os.Exit(1)
}
-}
-
-type mirumServer struct {
- pb.UnimplementedMirumServer
- secret []byte
- queue chan *pb.Task
- tasks sync.Map // task_id → taskMeta
- authedPeers sync.Map // peer addr string → true
-}
-
-func (s *mirumServer) Poll(ctx context.Context, req *pb.PollRequest) (*pb.Task, error) {
- select {
- case task := <-s.queue:
- slog.Info("task dispatched", "id", task.Id, "repo", task.RepoFullName)
- return task, nil
- case <-ctx.Done():
- return nil, ctx.Err()
- }
-}
-func (s *mirumServer) Complete(ctx context.Context, result *pb.TaskResult) (*pb.CompleteResponse, error) {
- meta, ok := s.tasks.LoadAndDelete(result.TaskId)
- if !ok {
- return nil, fmt.Errorf("unknown task: %s", result.TaskId)
- }
- m := meta.(taskMeta)
+ slog.Info("listening", "grpc", grpcLn.Addr(), "web", webLn.Addr())
- status := forges.StatusSuccess
- desc := "Build passed"
- if !result.Success {
- status = forges.StatusFailure
- desc = "Build failed"
- if result.Error != "" {
- desc = result.Error
+ go func() {
+ if err := wwwSrv.Serve(webLn); err != nil && err != http.ErrServerClosed {
+ slog.Error("web server failed", "err", err)
+ os.Exit(1)
}
- }
-
- if err := m.forge.SetStatus(ctx, m.event, status, desc); err != nil {
- slog.Error("set status", "task", result.TaskId, "err", err)
- }
-
- slog.Info("task complete", "id", result.TaskId, "success", result.Success)
- return &pb.CompleteResponse{}, nil
-}
-
-func (s *mirumServer) Handshake(stream pb.Mirum_HandshakeServer) error {
- hs := protocol.NewServerHandshake(s.secret)
-
- // Step 1: receive worker nonce
- in, err := stream.Recv()
- if err != nil {
- return fmt.Errorf("recv worker challenge: %w", err)
- }
- wc := in.GetWorkerChallenge()
- if wc == nil {
- return fmt.Errorf("expected WorkerChallenge")
- }
-
- // Step 2: send server nonce + proof
- serverNonce, proof, err := hs.Challenge(wc.GetNonce())
- if err != nil {
- return err
- }
- if err := stream.Send(&pb.HandshakeOut{
- Step: &pb.HandshakeOut_ServerChallenge{
- ServerChallenge: &pb.ServerChallenge{
- Nonce: serverNonce,
- Proof: proof,
- },
- },
- }); err != nil {
- return fmt.Errorf("send server challenge: %w", err)
- }
-
- // Step 3: receive worker proof + metadata
- in, err = stream.Recv()
- if err != nil {
- return fmt.Errorf("recv worker proof: %w", err)
- }
- wp := in.GetWorkerProof()
- if wp == nil {
- return fmt.Errorf("expected WorkerProof")
- }
-
- wt := wp.GetWorkerTime()
- if wt == nil {
- return s.reject(stream, "worker_time is required")
- }
-
- if err := hs.Verify(wp.GetProof(), wt.AsTime()); err != nil {
- return s.reject(stream, err.Error())
- }
-
- var warnings []string
- if skew := time.Since(wt.AsTime()).Abs(); skew > 10*time.Second {
- warnings = append(warnings, fmt.Sprintf("clock drift: %s", skew.Truncate(time.Second)))
- }
-
- slog.Info("worker connected",
- "id", fmt.Sprintf("%x", wp.GetId()),
- "name", wp.GetName(),
- "os", wp.GetOs(),
- "arch", wp.GetArch(),
- "runtime", wp.GetRuntime(),
- )
-
- // Step 4: accept
- if p, ok := peer.FromContext(stream.Context()); ok {
- s.authedPeers.Store(p.Addr.String(), true)
- }
- return s.sendResult(stream, nil, warnings)
-}
-
-func (s *mirumServer) sendResult(stream pb.Mirum_HandshakeServer, errMsg *string, warnings []string) error {
- return stream.Send(&pb.HandshakeOut{
- Step: &pb.HandshakeOut_ServerResult{
- ServerResult: &pb.ServerResult{
- Error: errMsg,
- ServerVersion: protocol.VersionProto(),
- ServerTime: timestamppb.Now(),
- Warnings: warnings,
- },
- },
- })
-}
-
-func (s *mirumServer) reject(stream pb.Mirum_HandshakeServer, reason string) error {
- if err := s.sendResult(stream, &reason, nil); err != nil {
- return err
- }
- return fmt.Errorf("%s", reason)
-}
-
-func peerAddr(ctx context.Context) string {
- if p, ok := peer.FromContext(ctx); ok {
- return p.Addr.String()
- }
- return ""
-}
-
-func (s *mirumServer) requireAuth(ctx context.Context, method string) error {
- if method == pb.Mirum_Handshake_FullMethodName {
- return nil
- }
- if _, ok := s.authedPeers.Load(peerAddr(ctx)); !ok {
- return status.Error(codes.Unauthenticated, "handshake required")
- }
- return nil
-}
-
-func (s *mirumServer) unaryInterceptor(ctx context.Context, req any, info *grpc.UnaryServerInfo, handler grpc.UnaryHandler) (any, error) {
- if err := s.requireAuth(ctx, info.FullMethod); err != nil {
- return nil, err
- }
- return handler(ctx, req)
-}
-
-func (s *mirumServer) streamInterceptor(srv any, ss grpc.ServerStream, info *grpc.StreamServerInfo, handler grpc.StreamHandler) error {
- if err := s.requireAuth(ss.Context(), info.FullMethod); err != nil {
- return err
- }
- if info.FullMethod == pb.Mirum_Handshake_FullMethodName {
- done := make(chan error, 1)
- go func() { done <- handler(srv, ss) }()
- select {
- case err := <-done:
- return err
- case <-time.After(30 * time.Second):
- return fmt.Errorf("handshake timeout")
+ }()
+ go func() {
+ if err := grpcSrv.Serve(grpcLn); err != nil {
+ slog.Error("grpc server failed", "err", err)
+ os.Exit(1)
}
- }
- return handler(srv, ss)
-}
+ }()
-// connTracker implements stats.Handler to clean up authedPeers on disconnect.
-type connTracker struct {
- stats.Handler
- server *mirumServer
-}
+ sup.Ready()
+ go sup.StartWatchdog(ctx)
-func (t *connTracker) TagConn(ctx context.Context, info *stats.ConnTagInfo) context.Context {
- return ctx
-}
+ <-ctx.Done()
+ slog.Info("shutting down")
+ sup.Stopping()
-func (t *connTracker) TagRPC(ctx context.Context, info *stats.RPCTagInfo) context.Context {
- return ctx
-}
+ // Hard deadline: if graceful shutdown takes too long, exit.
+ time.AfterFunc(30*time.Second, func() {
+ slog.Error("shutdown timed out, forcing exit")
+ os.Exit(1)
+ })
-func (t *connTracker) HandleRPC(ctx context.Context, s stats.RPCStats) {}
+ srv.Close()
-func (t *connTracker) HandleConn(ctx context.Context, s stats.ConnStats) {
- if _, ok := s.(*stats.ConnEnd); ok {
- addr := peerAddr(ctx)
- if addr != "" {
- t.server.authedPeers.Delete(addr)
- slog.Debug("peer disconnected", "addr", addr)
- }
- }
+ wwwSrv.Shutdown(ctx)
+ grpcSrv.GracefulStop()
}
// listeners returns gRPC and HTTP listeners.
// With systemd socket activation it expects two named fds: "grpc" and "http".
-// Without socket activation it falls back to cfg.GrpcAddr and cfg.WwwAddr.
-func listeners() (grpcLn, httpLn net.Listener, err error) {
+// Without socket activation it falls back to configured addresses.
+func listeners(cfg *config) (grpcLn, webLn net.Listener, err error) {
named, err := activation.ListenersWithNames()
if err != nil {
return nil, nil, fmt.Errorf("socket activation: %w", err)
@@ -378,24 +116,25 @@ func listeners() (grpcLn, httpLn net.Listener, err error) {
if lns := named["grpc"]; len(lns) > 0 {
grpcLn = lns[0]
+ } else if grpcLn, err = net.Listen("tcp", cfg.GrpcAddr); err != nil {
+ return nil, nil, err
}
- if lns := named["http"]; len(lns) > 0 {
- httpLn = lns[0]
- }
-
- if grpcLn == nil {
- grpcLn, err = net.Listen("tcp", cfg.GrpcAddr)
+ defer func() {
if err != nil {
- return nil, nil, err
+ _ = grpcLn.Close()
}
+ }()
+
+ if lns := named["web"]; len(lns) > 0 {
+ webLn = lns[0]
+ } else if webLn, err = net.Listen("tcp", cfg.WwwAddr); err != nil {
+ return nil, nil, err
}
- if httpLn == nil {
- httpLn, err = net.Listen("tcp", cfg.WwwAddr)
+ defer func() {
if err != nil {
- grpcLn.Close()
- return nil, nil, err
+ _ = webLn.Close()
}
- }
+ }()
- return grpcLn, httpLn, nil
+ return grpcLn, webLn, nil
}
diff --git a/cmd/mirumd/server.go b/cmd/mirumd/server.go
new file mode 100644
--- /dev/null
+++ b/cmd/mirumd/server.go
@@ -0,0 +1,76 @@
+// Copyright (c) 2026 Nikolay Govorov
+// SPDX-License-Identifier: AGPL-3.0-or-later
+
+package main
+
+import (
+ "context"
+ "fmt"
+ "log/slog"
+ "sync"
+ "sync/atomic"
+
+ "dimidiumlabs/mirum/internal/database"
+ "dimidiumlabs/mirum/internal/forges"
+ "dimidiumlabs/mirum/internal/protocol/pb"
+)
+
+// server holds the shared application state.
+type server struct {
+ cfg *config
+ db *database.DB
+ forge forges.Forge
+
+ queue chan *pb.Task
+ tasks sync.Map // task_id → *forges.PushEvent
+ taskCounter atomic.Int64
+}
+
+func (s *server) Close() {
+ close(s.queue)
+}
+
+func (s *server) enqueue(ev *forges.PushEvent) string {
+ s.taskCounter.Add(1)
+ id := fmt.Sprintf("task-%d", s.taskCounter.Load())
+
+ slog.Info("push", "repo", ev.Owner+"/"+ev.Repo, "branch", ev.Branch, "sha", ev.SHA[:8], "task", id)
+
+ s.tasks.Store(id, ev)
+ _ = s.forge.SetStatus(context.Background(), ev, forges.StatusPending, "Queued")
+
+ s.queue <- &pb.Task{
+ Id: id,
+ CloneUrl: s.forge.AuthURL(ev.CloneURL),
+ Branch: ev.Branch,
+ Sha: ev.SHA,
+ RepoFullName: ev.Owner + "/" + ev.Repo,
+ }
+
+ return id
+}
+
+func (s *server) complete(ctx context.Context, taskID string, success bool, errMsg string) error {
+ val, ok := s.tasks.LoadAndDelete(taskID)
+ if !ok {
+ return fmt.Errorf("unknown task: %s", taskID)
+ }
+ ev := val.(*forges.PushEvent)
+
+ st := forges.StatusSuccess
+ desc := "Build passed"
+ if !success {
+ st = forges.StatusFailure
+ desc = "Build failed"
+ if errMsg != "" {
+ desc = errMsg
+ }
+ }
+
+ if err := s.forge.SetStatus(ctx, ev, st, desc); err != nil {
+ slog.Error("set status", "task", taskID, "err", err)
+ }
+
+ slog.Info("task complete", "id", taskID, "success", success)
+ return nil
+}
diff --git a/cmd/mirumd/server_grpc.go b/cmd/mirumd/server_grpc.go
new file mode 100644
--- /dev/null
+++ b/cmd/mirumd/server_grpc.go
@@ -0,0 +1,222 @@
+// Copyright (c) 2026 Nikolay Govorov
+// SPDX-License-Identifier: AGPL-3.0-or-later
+
+package main
+
+import (
+ "context"
+ "fmt"
+ "log/slog"
+ "sync"
+ "time"
+
+ "dimidiumlabs/mirum/internal/protocol"
+ "dimidiumlabs/mirum/internal/protocol/pb"
+
+ "google.golang.org/grpc"
+ "google.golang.org/grpc/codes"
+ "google.golang.org/grpc/peer"
+ "google.golang.org/grpc/stats"
+ "google.golang.org/grpc/status"
+ "google.golang.org/protobuf/types/known/timestamppb"
+)
+
+func NewGrpcServer(ctx context.Context, srv *server, secret []byte) *grpc.Server {
+ gsrv := &grpcService{
+ ctx: ctx,
+ srv: srv,
+ secret: secret,
+ }
+ s := grpc.NewServer(
+ grpc.UnaryInterceptor(gsrv.unaryInterceptor),
+ grpc.StreamInterceptor(gsrv.streamInterceptor),
+ grpc.StatsHandler(&connTracker{gsrv: gsrv}),
+ )
+ pb.RegisterMirumServer(s, gsrv)
+ return s
+}
+
+// grpcService is the gRPC transport adapter over server.
+type grpcService struct {
+ pb.UnimplementedMirumServer
+
+ ctx context.Context
+ srv *server
+ secret []byte
+ authedPeers sync.Map // peer addr string → true
+}
+
+func (g *grpcService) Poll(ctx context.Context, req *pb.PollRequest) (*pb.Task, error) {
+ select {
+ case task, ok := <-g.srv.queue:
+ if !ok {
+ return nil, status.Error(codes.Unavailable, "server is shutting down")
+ }
+ slog.Info("task dispatched", "id", task.Id, "repo", task.RepoFullName)
+ return task, nil
+ case <-ctx.Done():
+ return nil, ctx.Err()
+ }
+}
+
+func (g *grpcService) Complete(ctx context.Context, result *pb.TaskResult) (*pb.CompleteResponse, error) {
+ if err := g.srv.complete(ctx, result.TaskId, result.Success, result.Error); err != nil {
+ return nil, err
+ }
+ return &pb.CompleteResponse{}, nil
+}
+
+func (g *grpcService) Handshake(stream pb.Mirum_HandshakeServer) error {
+ hs := protocol.NewServerHandshake(g.secret)
+
+ // Step 1: receive worker nonce
+ in, err := stream.Recv()
+ if err != nil {
+ return fmt.Errorf("recv worker challenge: %w", err)
+ }
+ wc := in.GetWorkerChallenge()
+ if wc == nil {
+ return fmt.Errorf("expected WorkerChallenge")
+ }
+
+ // Step 2: send server nonce + proof
+ serverNonce, proof, err := hs.Challenge(wc.GetNonce())
+ if err != nil {
+ return err
+ }
+ if err := stream.Send(&pb.HandshakeOut{
+ Step: &pb.HandshakeOut_ServerChallenge{
+ ServerChallenge: &pb.ServerChallenge{
+ Nonce: serverNonce,
+ Proof: proof,
+ },
+ },
+ }); err != nil {
+ return fmt.Errorf("send server challenge: %w", err)
+ }
+
+ // Step 3: receive worker proof + metadata
+ in, err = stream.Recv()
+ if err != nil {
+ return fmt.Errorf("recv worker proof: %w", err)
+ }
+ wp := in.GetWorkerProof()
+ if wp == nil {
+ return fmt.Errorf("expected WorkerProof")
+ }
+
+ wt := wp.GetWorkerTime()
+ if wt == nil {
+ return g.reject(stream, "worker_time is required")
+ }
+
+ if err := hs.Verify(wp.GetProof(), wt.AsTime()); err != nil {
+ return g.reject(stream, err.Error())
+ }
+
+ var warnings []string
+ if skew := time.Since(wt.AsTime()).Abs(); skew > 10*time.Second {
+ warnings = append(warnings, fmt.Sprintf("clock drift: %s", skew.Truncate(time.Second)))
+ }
+
+ slog.Info("worker connected",
+ "id", fmt.Sprintf("%x", wp.GetId()),
+ "name", wp.GetName(),
+ "os", wp.GetOs(),
+ "arch", wp.GetArch(),
+ "runtime", wp.GetRuntime(),
+ )
+
+ // Step 4: accept
+ if p, ok := peer.FromContext(stream.Context()); ok {
+ g.authedPeers.Store(p.Addr.String(), true)
+ }
+ return g.sendResult(stream, nil, warnings)
+}
+
+func (g *grpcService) sendResult(stream pb.Mirum_HandshakeServer, errMsg *string, warnings []string) error {
+ return stream.Send(&pb.HandshakeOut{
+ Step: &pb.HandshakeOut_ServerResult{
+ ServerResult: &pb.ServerResult{
+ Error: errMsg,
+ ServerVersion: protocol.VersionProto(),
+ ServerTime: timestamppb.Now(),
+ Warnings: warnings,
+ },
+ },
+ })
+}
+
+func (g *grpcService) reject(stream pb.Mirum_HandshakeServer, reason string) error {
+ if err := g.sendResult(stream, &reason, nil); err != nil {
+ return err
+ }
+ return fmt.Errorf("%s", reason)
+}
+
+func peerAddr(ctx context.Context) string {
+ if p, ok := peer.FromContext(ctx); ok {
+ return p.Addr.String()
+ }
+ return ""
+}
+
+func (g *grpcService) requireAuth(ctx context.Context, method string) error {
+ if method == pb.Mirum_Handshake_FullMethodName {
+ return nil
+ }
+ if _, ok := g.authedPeers.Load(peerAddr(ctx)); !ok {
+ return status.Error(codes.Unauthenticated, "handshake required")
+ }
+ return nil
+}
+
+func (g *grpcService) unaryInterceptor(ctx context.Context, req any, info *grpc.UnaryServerInfo, handler grpc.UnaryHandler) (any, error) {
+ if err := g.requireAuth(ctx, info.FullMethod); err != nil {
+ return nil, err
+ }
+ return handler(ctx, req)
+}
+
+func (g *grpcService) streamInterceptor(srv any, ss grpc.ServerStream, info *grpc.StreamServerInfo, handler grpc.StreamHandler) error {
+ if err := g.requireAuth(ss.Context(), info.FullMethod); err != nil {
+ return err
+ }
+ if info.FullMethod == pb.Mirum_Handshake_FullMethodName {
+ done := make(chan error, 1)
+ go func() { done <- handler(srv, ss) }()
+ select {
+ case err := <-done:
+ return err
+ case <-time.After(30 * time.Second):
+ return fmt.Errorf("handshake timeout")
+ }
+ }
+ return handler(srv, ss)
+}
+
+// connTracker implements stats.Handler to clean up authedPeers on disconnect.
+type connTracker struct {
+ stats.Handler
+ gsrv *grpcService
+}
+
+func (t *connTracker) TagConn(ctx context.Context, info *stats.ConnTagInfo) context.Context {
+ return ctx
+}
+
+func (t *connTracker) TagRPC(ctx context.Context, info *stats.RPCTagInfo) context.Context {
+ return ctx
+}
+
+func (t *connTracker) HandleRPC(ctx context.Context, s stats.RPCStats) {}
+
+func (t *connTracker) HandleConn(ctx context.Context, s stats.ConnStats) {
+ if _, ok := s.(*stats.ConnEnd); ok {
+ addr := peerAddr(ctx)
+ if addr != "" {
+ t.gsrv.authedPeers.Delete(addr)
+ slog.Debug("peer disconnected", "addr", addr)
+ }
+ }
+}
diff --git a/cmd/mirumd/server_web.go b/cmd/mirumd/server_web.go
new file mode 100644
--- /dev/null
+++ b/cmd/mirumd/server_web.go
@@ -0,0 +1,60 @@
+// Copyright (c) 2026 Nikolay Govorov
+// SPDX-License-Identifier: AGPL-3.0-or-later
+
+package main
+
+import (
+ "context"
+ "errors"
+ "fmt"
+ "io"
+ "net"
+ "net/http"
+
+ "dimidiumlabs/mirum/internal/forges"
+)
+
+func NewWwwServer(ctx context.Context, srv *server) *http.Server {
+ return &http.Server{
+ Handler: wwwRoutes(srv),
+ BaseContext: func(_ net.Listener) context.Context {
+ return ctx
+ },
+ }
+}
+
+func wwwRoutes(srv *server) http.Handler {
+ mux := http.NewServeMux()
+
+ mux.HandleFunc("GET /", func(w http.ResponseWriter, r *http.Request) {
+ w.Header().Set("Content-Type", "text/html; charset=utf-8")
+ fmt.Fprint(w, `<!DOCTYPE html><html><head><meta charset="utf-8"><title>Mirum</title></head><body><h1>Mirum</h1><p>CI server is running.</p></body></html>`)
+ })
+
+ mux.HandleFunc("POST /webhook", func(w http.ResponseWriter, r *http.Request) {
+ body, err := io.ReadAll(r.Body)
+ if err != nil {
+ http.Error(w, "read body", http.StatusBadRequest)
+ return
+ }
+
+ ev, err := srv.forge.Webhook(r, body)
+ if errors.Is(err, forges.ErrInvalidSignature) {
+ http.Error(w, "invalid signature", http.StatusUnauthorized)
+ return
+ }
+ if err != nil {
+ http.Error(w, err.Error(), http.StatusBadRequest)
+ return
+ }
+ if ev == nil {
+ w.WriteHeader(http.StatusNoContent)
+ return
+ }
+
+ srv.enqueue(ev)
+ w.WriteHeader(http.StatusAccepted)
+ })
+
+ return mux
+}
diff --git a/cmd/mirumw/main.go b/cmd/mirumw/main.go
index 40d3837..40b5f2b 100644
--- a/cmd/mirumw/main.go
+++ b/cmd/mirumw/main.go
@@ -28,7 +28,7 @@ func main() {
ctx := sup.WaitForStop(context.Background())
sup.Ready()
- go sup.StartWatchdog()
+ go sup.StartWatchdog(ctx)
backoff := protocol.NewBackoff()
diff --git a/go.mod b/go.mod
index ae3cb3b..13f91a3 100644
--- a/go.mod
+++ b/go.mod
@@ -4,15 +4,32 @@ go 1.26.1
require (
github.com/coreos/go-systemd/v22 v22.7.0
+ github.com/jackc/pgx/v5 v5.9.1
+ github.com/jackc/tern/v2 v2.3.6
go.starlark.net v0.0.0-20260326113308-fadfc96def35
+ google.golang.org/grpc v1.79.3
+ google.golang.org/protobuf v1.36.11
gopkg.in/yaml.v3 v3.0.1
)
require (
+ dario.cat/mergo v1.0.1 // indirect
+ github.com/Masterminds/goutils v1.1.1 // indirect
+ github.com/Masterminds/semver/v3 v3.3.0 // indirect
+ github.com/Masterminds/sprig/v3 v3.3.0 // indirect
+ github.com/google/uuid v1.6.0 // indirect
+ github.com/huandu/xstrings v1.5.0 // indirect
+ github.com/jackc/pgpassfile v1.0.0 // indirect
+ github.com/jackc/pgservicefile v0.0.0-20240606120523-5a60cdf6a761 // indirect
+ github.com/jackc/puddle/v2 v2.2.2 // indirect
+ github.com/mitchellh/copystructure v1.2.0 // indirect
+ github.com/mitchellh/reflectwalk v1.0.2 // indirect
+ github.com/shopspring/decimal v1.4.0 // indirect
+ github.com/spf13/cast v1.7.0 // indirect
+ golang.org/x/crypto v0.49.0 // indirect
golang.org/x/net v0.52.0 // indirect
+ golang.org/x/sync v0.20.0 // indirect
golang.org/x/sys v0.42.0 // indirect
golang.org/x/text v0.35.0 // indirect
google.golang.org/genproto/googleapis/rpc v0.0.0-20260330182312-d5a96adf58d8 // indirect
- google.golang.org/grpc v1.79.3 // indirect
- google.golang.org/protobuf v1.36.11 // indirect
)
diff --git a/go.sum b/go.sum
index 0d9449e..9d587a7 100644
--- a/go.sum
+++ b/go.sum
@@ -1,28 +1,98 @@
+dario.cat/mergo v1.0.1 h1:Ra4+bf83h2ztPIQYNP99R6m+Y7KfnARDfID+a+vLl4s=
+dario.cat/mergo v1.0.1/go.mod h1:uNxQE+84aUszobStD9th8a29P2fMDhsBdgRYvZOxGmk=
+github.com/Masterminds/goutils v1.1.1 h1:5nUrii3FMTL5diU80unEVvNevw1nH4+ZV4DSLVJLSYI=
+github.com/Masterminds/goutils v1.1.1/go.mod h1:8cTjp+g8YejhMuvIA5y2vz3BpJxksy863GQaJW2MFNU=
+github.com/Masterminds/semver/v3 v3.3.0 h1:B8LGeaivUe71a5qox1ICM/JLl0NqZSW5CHyL+hmvYS0=
+github.com/Masterminds/semver/v3 v3.3.0/go.mod h1:4V+yj/TJE1HU9XfppCwVMZq3I84lprf4nC11bSS5beM=
+github.com/Masterminds/sprig/v3 v3.3.0 h1:mQh0Yrg1XPo6vjYXgtf5OtijNAKJRNcTdOOGZe3tPhs=
+github.com/Masterminds/sprig/v3 v3.3.0/go.mod h1:Zy1iXRYNqNLUolqCpL4uhk6SHUMAOSCzdgBfDb35Lz0=
+github.com/cespare/xxhash/v2 v2.3.0 h1:UL815xU9SqsFlibzuggzjXhog7bL6oX9BbNZnL2UFvs=
+github.com/cespare/xxhash/v2 v2.3.0/go.mod h1:VGX0DQ3Q6kWi7AoAeZDth3/j3BFtOZR5XLFGgcrjCOs=
github.com/coreos/go-systemd/v22 v22.7.0 h1:LAEzFkke61DFROc7zNLX/WA2i5J8gYqe0rSj9KI28KA=
github.com/coreos/go-systemd/v22 v22.7.0/go.mod h1:xNUYtjHu2EDXbsxz1i41wouACIwT7Ybq9o0BQhMwD0w=
+github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
+github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c=
+github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
+github.com/frankban/quicktest v1.14.6 h1:7Xjx+VpznH+oBnejlPUj8oUpdxnVs4f8XU8WnHkI4W8=
+github.com/frankban/quicktest v1.14.6/go.mod h1:4ptaffx2x8+WTWXmUCuVU6aPUX1/Mz7zb5vbUoiM6w0=
+github.com/go-logr/logr v1.4.3 h1:CjnDlHq8ikf6E492q6eKboGOC0T8CDaOvkHCIg8idEI=
+github.com/go-logr/logr v1.4.3/go.mod h1:9T104GzyrTigFIr8wt5mBrctHMim0Nb2HLGrmQ40KvY=
+github.com/go-logr/stdr v1.2.2 h1:hSWxHoqTgW2S2qGc0LTAI563KZ5YKYRhT3MFKZMbjag=
+github.com/go-logr/stdr v1.2.2/go.mod h1:mMo/vtBO5dYbehREoey6XUKy/eSumjCCveDpRre4VKE=
+github.com/golang/protobuf v1.5.4 h1:i7eJL8qZTpSEXOPTxNKhASYpMn+8e5Q6AdndVa1dWek=
+github.com/golang/protobuf v1.5.4/go.mod h1:lnTiLA8Wa4RWRcIUkrtSVa5nRhsEGBg48fD6rSs7xps=
github.com/google/go-cmp v0.7.0 h1:wk8382ETsv4JYUZwIsn6YpYiWiBsYLSJiTsyBybVuN8=
github.com/google/go-cmp v0.7.0/go.mod h1:pXiqmnSA92OHEEa9HXL2W4E7lf9JzCmGVUdgjX3N/iU=
+github.com/google/uuid v1.6.0 h1:NIvaJDMOsjHA8n1jAhLSgzrAzy1Hgr+hNrb57e+94F0=
+github.com/google/uuid v1.6.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo=
+github.com/huandu/xstrings v1.5.0 h1:2ag3IFq9ZDANvthTwTiqSSZLjDc+BedvHPAp5tJy2TI=
+github.com/huandu/xstrings v1.5.0/go.mod h1:y5/lhBue+AyNmUVz9RLU9xbLR0o4KIIExikq4ovT0aE=
+github.com/jackc/pgpassfile v1.0.0 h1:/6Hmqy13Ss2zCq62VdNG8tM1wchn8zjSGOBJ6icpsIM=
+github.com/jackc/pgpassfile v1.0.0/go.mod h1:CEx0iS5ambNFdcRtxPj5JhEz+xB6uRky5eyVu/W2HEg=
+github.com/jackc/pgservicefile v0.0.0-20240606120523-5a60cdf6a761 h1:iCEnooe7UlwOQYpKFhBabPMi4aNAfoODPEFNiAnClxo=
+github.com/jackc/pgservicefile v0.0.0-20240606120523-5a60cdf6a761/go.mod h1:5TJZWKEWniPve33vlWYSoGYefn3gLQRzjfDlhSJ9ZKM=
+github.com/jackc/pgx/v5 v5.9.1 h1:uwrxJXBnx76nyISkhr33kQLlUqjv7et7b9FjCen/tdc=
+github.com/jackc/pgx/v5 v5.9.1/go.mod h1:mal1tBGAFfLHvZzaYh77YS/eC6IX9OWbRV1QIIM0Jn4=
+github.com/jackc/puddle/v2 v2.2.2 h1:PR8nw+E/1w0GLuRFSmiioY6UooMp6KJv0/61nB7icHo=
+github.com/jackc/puddle/v2 v2.2.2/go.mod h1:vriiEXHvEE654aYKXXjOvZM39qJ0q+azkZFrfEOc3H4=
+github.com/jackc/tern/v2 v2.3.6 h1:sqBIZ/CBtfMLz7zdUof0N6cVUBRVGBZ7S+F2OdCp9XU=
+github.com/jackc/tern/v2 v2.3.6/go.mod h1:SrtwsdBRKkeTOjuLd6ISNqaLOtaLX+jOTLrpP+lJQe0=
+github.com/kr/pretty v0.3.1 h1:flRD4NNwYAUpkphVc1HcthR4KEIFJ65n8Mw5qdRn3LE=
+github.com/kr/pretty v0.3.1/go.mod h1:hoEshYVHaxMs3cyo3Yncou5ZscifuDolrwPKZanG3xk=
+github.com/kr/text v0.2.0 h1:5Nx0Ya0ZqY2ygV366QzturHI13Jq95ApcVaJBhpS+AY=
+github.com/kr/text v0.2.0/go.mod h1:eLer722TekiGuMkidMxC/pM04lWEeraHUUmBw8l2grE=
+github.com/mitchellh/copystructure v1.2.0 h1:vpKXTN4ewci03Vljg/q9QvCGUDttBOGBIa15WveJJGw=
+github.com/mitchellh/copystructure v1.2.0/go.mod h1:qLl+cE2AmVv+CoeAwDPye/v+N2HKCj9FbZEVFJRxO9s=
+github.com/mitchellh/reflectwalk v1.0.2 h1:G2LzWKi524PWgd3mLHV8Y5k7s6XUvT0Gef6zxSIeXaQ=
+github.com/mitchellh/reflectwalk v1.0.2/go.mod h1:mSTlrgnPZtwu0c4WaC2kGObEpuNDbx0jmZXqmk4esnw=
+github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM=
+github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
+github.com/rogpeppe/go-internal v1.9.0 h1:73kH8U+JUqXU8lRuOHeVHaa/SZPifC7BkcraZVejAe8=
+github.com/rogpeppe/go-internal v1.9.0/go.mod h1:WtVeX8xhTBvf0smdhujwtBcq4Qrzq/fJaraNFVN+nFs=
+github.com/shopspring/decimal v1.4.0 h1:bxl37RwXBklmTi0C79JfXCEBD1cqqHt0bbgBAGFp81k=
+github.com/shopspring/decimal v1.4.0/go.mod h1:gawqmDU56v4yIKSwfBSFip1HdCCXN8/+DMd9qYNcwME=
+github.com/spf13/cast v1.7.0 h1:ntdiHjuueXFgm5nzDRdOS4yfT43P5Fnud6DH50rz/7w=
+github.com/spf13/cast v1.7.0/go.mod h1:ancEpBxwJDODSW/UG4rDrAqiKolqNNh2DX3mk86cAdo=
+github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME=
+github.com/stretchr/testify v1.3.0/go.mod h1:M5WIy9Dh21IEIfnGCwXGc5bZfKNJtfHm1UVUgZn+9EI=
+github.com/stretchr/testify v1.7.0/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg=
+github.com/stretchr/testify v1.11.1 h1:7s2iGBzp5EwR7/aIZr8ao5+dra3wiQyKjjFuvgVKu7U=
+github.com/stretchr/testify v1.11.1/go.mod h1:wZwfW3scLgRK+23gO65QZefKpKQRnfz6sD981Nm4B6U=
+go.opentelemetry.io/auto/sdk v1.2.1 h1:jXsnJ4Lmnqd11kwkBV2LgLoFMZKizbCi5fNZ/ipaZ64=
+go.opentelemetry.io/auto/sdk v1.2.1/go.mod h1:KRTj+aOaElaLi+wW1kO/DZRXwkF4C5xPbEe3ZiIhN7Y=
+go.opentelemetry.io/otel v1.39.0 h1:8yPrr/S0ND9QEfTfdP9V+SiwT4E0G7Y5MO7p85nis48=
+go.opentelemetry.io/otel v1.39.0/go.mod h1:kLlFTywNWrFyEdH0oj2xK0bFYZtHRYUdv1NklR/tgc8=
+go.opentelemetry.io/otel/metric v1.39.0 h1:d1UzonvEZriVfpNKEVmHXbdf909uGTOQjA0HF0Ls5Q0=
+go.opentelemetry.io/otel/metric v1.39.0/go.mod h1:jrZSWL33sD7bBxg1xjrqyDjnuzTUB0x1nBERXd7Ftcs=
+go.opentelemetry.io/otel/sdk v1.39.0 h1:nMLYcjVsvdui1B/4FRkwjzoRVsMK8uL/cj0OyhKzt18=
+go.opentelemetry.io/otel/sdk v1.39.0/go.mod h1:vDojkC4/jsTJsE+kh+LXYQlbL8CgrEcwmt1ENZszdJE=
+go.opentelemetry.io/otel/sdk/metric v1.39.0 h1:cXMVVFVgsIf2YL6QkRF4Urbr/aMInf+2WKg+sEJTtB8=
+go.opentelemetry.io/otel/sdk/metric v1.39.0/go.mod h1:xq9HEVH7qeX69/JnwEfp6fVq5wosJsY1mt4lLfYdVew=
+go.opentelemetry.io/otel/trace v1.39.0 h1:2d2vfpEDmCJ5zVYz7ijaJdOF59xLomrvj7bjt6/qCJI=
+go.opentelemetry.io/otel/trace v1.39.0/go.mod h1:88w4/PnZSazkGzz/w84VHpQafiU4EtqqlVdxWy+rNOA=
go.starlark.net v0.0.0-20260326113308-fadfc96def35 h1:VYAqieSOJNxBDX8KJneTAwvdf4J4zRDE2u+UFXtt9h4=
go.starlark.net v0.0.0-20260326113308-fadfc96def35/go.mod h1:Iue6g6iirlfLoVi/DYCi5/x0h/bAOuWF3dULTKpt2Vo=
-golang.org/x/net v0.48.0 h1:zyQRTTrjc33Lhh0fBgT/H3oZq9WuvRR5gPC70xpDiQU=
-golang.org/x/net v0.48.0/go.mod h1:+ndRgGjkh8FGtu1w1FGbEC31if4VrNVMuKTgcAAnQRY=
+golang.org/x/crypto v0.49.0 h1:+Ng2ULVvLHnJ/ZFEq4KdcDd/cfjrrjjNSXNzxg0Y4U4=
+golang.org/x/crypto v0.49.0/go.mod h1:ErX4dUh2UM+CFYiXZRTcMpEcN8b/1gxEuv3nODoYtCA=
golang.org/x/net v0.52.0 h1:He/TN1l0e4mmR3QqHMT2Xab3Aj3L9qjbhRm78/6jrW0=
golang.org/x/net v0.52.0/go.mod h1:R1MAz7uMZxVMualyPXb+VaqGSa3LIaUqk0eEt3w36Sw=
+golang.org/x/sync v0.20.0 h1:e0PTpb7pjO8GAtTs2dQ6jYa5BWYlMuX047Dco/pItO4=
+golang.org/x/sync v0.20.0/go.mod h1:9xrNwdLfx4jkKbNva9FpL6vEN7evnE43NNNJQ2LF3+0=
golang.org/x/sys v0.42.0 h1:omrd2nAlyT5ESRdCLYdm3+fMfNFE/+Rf4bDIQImRJeo=
golang.org/x/sys v0.42.0/go.mod h1:4GL1E5IUh+htKOUEOaiffhrAeqysfVGipDYzABqnCmw=
-golang.org/x/text v0.32.0 h1:ZD01bjUt1FQ9WJ0ClOL5vxgxOI/sVCNgX1YtKwcY0mU=
-golang.org/x/text v0.32.0/go.mod h1:o/rUWzghvpD5TXrTIBuJU77MTaN0ljMWE47kxGJQ7jY=
golang.org/x/text v0.35.0 h1:JOVx6vVDFokkpaq1AEptVzLTpDe9KGpj5tR4/X+ybL8=
golang.org/x/text v0.35.0/go.mod h1:khi/HExzZJ2pGnjenulevKNX1W67CUy0AsXcNubPGCA=
-google.golang.org/genproto/googleapis/rpc v0.0.0-20251202230838-ff82c1b0f217 h1:gRkg/vSppuSQoDjxyiGfN4Upv/h/DQmIR10ZU8dh4Ww=
-google.golang.org/genproto/googleapis/rpc v0.0.0-20251202230838-ff82c1b0f217/go.mod h1:7i2o+ce6H/6BluujYR+kqX3GKH+dChPTQU19wjRPiGk=
+gonum.org/v1/gonum v0.16.0 h1:5+ul4Swaf3ESvrOnidPp4GZbzf0mxVQpDCYUQE7OJfk=
+gonum.org/v1/gonum v0.16.0/go.mod h1:fef3am4MQ93R2HHpKnLk4/Tbh/s0+wqD5nfa6Pnwy4E=
google.golang.org/genproto/googleapis/rpc v0.0.0-20260330182312-d5a96adf58d8 h1:OHkuo1i98/05rzpm9NBbfEtpJH/k3abEgZUKaAuCI7Y=
google.golang.org/genproto/googleapis/rpc v0.0.0-20260330182312-d5a96adf58d8/go.mod h1:4Hqkh8ycfw05ld/3BWL7rJOSfebL2Q+DVDeRgYgxUU8=
google.golang.org/grpc v1.79.3 h1:sybAEdRIEtvcD68Gx7dmnwjZKlyfuc61Dyo9pGXXkKE=
google.golang.org/grpc v1.79.3/go.mod h1:KmT0Kjez+0dde/v2j9vzwoAScgEPx/Bw1CYChhHLrHQ=
google.golang.org/protobuf v1.36.11 h1:fV6ZwhNocDyBLK0dj+fg8ektcVegBBuEolpbTQyBNVE=
google.golang.org/protobuf v1.36.11/go.mod h1:HTf+CrKn2C3g5S8VImy6tdcUvCska2kB7j23XfzDpco=
-gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405 h1:yhCVgyC4o1eVCa2tZl7eS0r+SDo693bJlVdllGtEeKM=
gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
+gopkg.in/check.v1 v1.0.0-20201130134442-10cb98267c6c h1:Hei/4ADfdWqJk1ZMxUNpqntNwaWcugrBjAiHlqqRiVk=
+gopkg.in/check.v1 v1.0.0-20201130134442-10cb98267c6c/go.mod h1:JHkPIbrfpd72SG/EVd6muEfDQjcINNoR0C8j2r3qZ4Q=
+gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=
gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA=
gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=
diff --git a/internal/database/database.go b/internal/database/database.go
new file mode 100644
--- /dev/null
+++ b/internal/database/database.go
@@ -0,0 +1,68 @@
+// Copyright (c) 2026 Nikolay Govorov
+// SPDX-License-Identifier: AGPL-3.0-or-later
+
+package database
+
+import (
+ "context"
+ "errors"
+
+ "github.com/jackc/pgx/v5/pgxpool"
+ "github.com/jackc/tern/v2/migrate"
+)
+
+var (
+ errOpen = errors.New("database: failed to open")
+ errPing = errors.New("database: failed to ping")
+ errAcquire = errors.New("database: failed to acquire connection")
+ errMigrate = errors.New("database: failed to create migrator")
+)
+
+// DB wraps a pgx connection pool.
+type DB struct {
+ Pool *pgxpool.Pool
+}
+
+// Open connects to PostgreSQL and returns a DB.
+func Open(ctx context.Context, dsn string) (*DB, error) {
+ pool, err := pgxpool.New(ctx, dsn)
+ if err != nil {
+ return nil, errors.Join(errOpen, err)
+ }
+ if err := pool.Ping(ctx); err != nil {
+ pool.Close()
+ return nil, errors.Join(errPing, err)
+ }
+ return &DB{Pool: pool}, nil
+}
+
+// Close closes the connection pool.
+func (db *DB) Close() {
+ db.Pool.Close()
+}
+
+// Migrate applies all pending migrations.
+func (db *DB) Migrate(ctx context.Context) error {
+ conn, err := db.Pool.Acquire(ctx)
+ if err != nil {
+ return errors.Join(errAcquire, err)
+ }
+ defer conn.Release()
+
+ migrator, err := migrate.NewMigrator(ctx, conn.Conn(), "schema_version")
+ if err != nil {
+ return errors.Join(errMigrate, err)
+ }
+
+ migrator.AppendMigration("create_users",
+ `CREATE TABLE users (
+ id UUID PRIMARY KEY DEFAULT uuidv7(),
+ email TEXT NOT NULL UNIQUE,
+ password TEXT NOT NULL,
+ created_at TIMESTAMPTZ NOT NULL DEFAULT now()
+ )`,
+ `DROP TABLE users`,
+ )
+
+ return migrator.Migrate(ctx)
+}
diff --git a/internal/supervisor/supervisor.go b/internal/supervisor/supervisor.go
index 6005dd5..202644f 100644
--- a/internal/supervisor/supervisor.go
+++ b/internal/supervisor/supervisor.go
@@ -23,9 +23,9 @@ type Supervisor interface {
Stopping()
// StartWatchdog begins sending periodic keepalive pings.
- // Blocks forever; call as a goroutine. Returns immediately if the
- // supervisor does not require keepalives.
- StartWatchdog()
+ // Blocks until ctx is cancelled; call as a goroutine. Returns
+ // immediately if the supervisor does not require keepalives.
+ StartWatchdog(ctx context.Context)
// WaitForStop blocks until the supervisor or OS requests shutdown.
WaitForStop(ctx context.Context) context.Context
@@ -45,7 +45,7 @@ func (*noop) Ready() {}
func (*noop) Stopping() {}
-func (*noop) StartWatchdog() {}
+func (*noop) StartWatchdog(context.Context) {}
func (*noop) WaitForStop(ctx context.Context) context.Context {
ctx, stop := signal.NotifyContext(ctx, syscall.SIGINT, syscall.SIGTERM)
diff --git a/internal/supervisor/systemd.go b/internal/supervisor/systemd.go
index c4d6ff4..63c1c0d 100644
--- a/internal/supervisor/systemd.go
+++ b/internal/supervisor/systemd.go
@@ -39,7 +39,7 @@ func (*systemd) Stopping() {
}
}
-func (*systemd) StartWatchdog() {
+func (*systemd) StartWatchdog(ctx context.Context) {
usecStr := os.Getenv("WATCHDOG_USEC")
if usecStr == "" {
return
@@ -51,12 +51,19 @@ func (*systemd) StartWatchdog() {
}
interval := time.Duration(usec) * time.Microsecond / 2
+ ticker := time.NewTicker(interval)
+ defer ticker.Stop()
+
for {
if _, err := daemon.SdNotify(false, daemon.SdNotifyWatchdog); err != nil {
slog.Warn("sd_notify watchdog", "err", err)
return
}
- time.Sleep(interval)
+ select {
+ case <-ticker.C:
+ case <-ctx.Done():
+ return
+ }
}
}
diff --git a/pkg/mirumd.service b/pkg/mirumd.service
index 89051ba..e470c99 100644
--- a/pkg/mirumd.service
+++ b/pkg/mirumd.service
@@ -8,11 +8,11 @@ After=time-sync.target network-online.target remote-fs.target nss-lookup.target
Wants=time-sync.target
# Socket activation (optional):
-# Create mirumd.socket with named file descriptors "http" and "grpc":
+# Create mirumd.socket with named file descriptors "web" and "grpc":
#
# [Socket]
# ListenStream=0.0.0.0:3000
-# FileDescriptorName=http
+# FileDescriptorName=web
#
# [Socket]
# ListenStream=0.0.0.0:2026
diff --git a/pkg/mirumd.yaml b/pkg/mirumd.yaml
index c2cd3ea..c3041b2 100644
--- a/pkg/mirumd.yaml
+++ b/pkg/mirumd.yaml
@@ -2,9 +2,10 @@
# SPDX-License-Identifier: AGPL-3.0-or-later
# Ignored when the corresponding systemd socket activation fd is present.
-# See mirumd.socket for details (FileDescriptorName=grpc / http).
+# See mirumd.socket for details (FileDescriptorName=grpc / web).
grpc_addr: :2026
www_addr: :3000
secret: ""
token: ""
+dsn: ""
script: .mirum/main.star
generated by Gilti