mesh-delivery/modules/mesh-delivery-watchman/internal/socket/socket.go

// Package socket is how the watchman's tools ask its program what only the program reads (novox/hq ADR 0564 §12, as the
// view's tools ask the view, mesh-delivery-view internal/view/serve.go): DEAD_LETTERS is a stream, which the tool runtime
// grants no bundle, and resync is an act of the program. One line of JSON each way, on a unix socket on the program's
// machine; the socket keeps nothing.
package socket

import (
	"bufio"
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"io"
	"net"
	"os"
	"path/filepath"
	"strings"
	"time"
)

// Path is where the program listens unless MESH_WATCHMAN_SOCKET says otherwise.
const Path = "/var/lib/mesh-delivery-watchman-tools/tools.sock"

// PathOf is the socket's path as getenv says it.
func PathOf(getenv func(string) string) string {
	if p := strings.TrimSpace(getenv("MESH_WATCHMAN_SOCKET")); p != "" {
		return p
	}
	return Path
}

// The asks the program answers.
const (
	AskDeadLetters = "dead-letters"
	AskResync      = "resync"
)

// Ask is one ask of the tools: dead-letters (consumer, limit, id; deliver or drop with why) or resync.
type Ask struct {
	Ask      string `json:"ask"`
	Consumer string `json:"consumer,omitempty"`
	Limit    int    `json:"limit,omitempty"`
	ID       uint64 `json:"id,omitempty"`
	Act      string `json:"act,omitempty"`
	Why      string `json:"why,omitempty"`
}

// Listen listens on the unix socket at path, in a directory of mode 0755, the socket 0666.
func Listen(path string) (net.Listener, error) {
	dir := filepath.Dir(path)
	if err := os.MkdirAll(dir, 0o755); err != nil {
		return nil, err
	}
	if err := os.Remove(path); err != nil && !errors.Is(err, os.ErrNotExist) {
		return nil, err
	}
	l, err := net.Listen("unix", path)
	if err != nil {
		return nil, err
	}
	if err := os.Chmod(path, 0o666); err != nil {
		l.Close()
		return nil, err
	}
	return l, nil
}

// Serve answers each ask on l with answer until ctx ends.
func Serve(ctx context.Context, l net.Listener, answer func(Ask) (any, error)) error {
	go func() { <-ctx.Done(); l.Close() }()
	for {
		c, err := l.Accept()
		if err != nil {
			if ctx.Err() != nil {
				return nil
			}
			return err
		}
		go func() {
			defer c.Close()
			_ = c.SetDeadline(time.Now().Add(30 * time.Second))
			line, err := bufio.NewReader(io.LimitReader(c, 4096)).ReadBytes('\n')
			if err != nil && len(line) == 0 {
				return
			}
			var a Ask
			enc := json.NewEncoder(c)
			if json.Unmarshal(line, &a) != nil {
				_ = enc.Encode(map[string]any{"error": "refused: an ask is one line of JSON"})
				return
			}
			out, err := answer(a)
			if err != nil {
				_ = enc.Encode(map[string]any{"error": err.Error()})
				return
			}
			_ = enc.Encode(map[string]any{"answer": out})
		}()
	}
}

// On asks the program at path and returns its answer.
func On(path string, a Ask) (json.RawMessage, error) {
	c, err := net.DialTimeout("unix", path, 5*time.Second)
	if err != nil {
		return nil, fmt.Errorf("cannot reach the watchman at %s: %w", path, err)
	}
	defer c.Close()
	_ = c.SetDeadline(time.Now().Add(30 * time.Second))
	line, _ := json.Marshal(a)
	if _, err := c.Write(append(line, '\n')); err != nil {
		return nil, err
	}
	var reply struct {
		Answer json.RawMessage `json:"answer"`
		Error  string          `json:"error"`
	}
	if err := json.NewDecoder(c).Decode(&reply); err != nil {
		return nil, fmt.Errorf("the watchman's answer cannot be read: %w", err)
	}
	if reply.Error != "" {
		return nil, errors.New(reply.Error)
	}
	return reply.Answer, nil
}