1 files changed, 522 insertions, 0 deletions
diff --git a/cmd/mbf/internal/ci/ci.go b/cmd/mbf/internal/ci/ci.go
new file mode 100644
index 00000000..c9d585d4
--- /dev/null
+++ b/cmd/mbf/internal/ci/ci.go
@@ -0,0 +1,522 @@
+// Package ci implements the CI service and client.
+package ci
+
+import (
+	"archive/tar"
+	"bytes"
+	"compress/gzip"
+	"context"
+	"crypto/ed25519"
+	"encoding/binary"
+	"errors"
+	"io"
+	"io/fs"
+	"log"
+	"net"
+	"net/http"
+	"os"
+	"path"
+	"path/filepath"
+	"sync"
+	"syscall"
+	"time"
+	"unique"
+	"unsafe"
+	_ "unsafe" // for go:linkname
+
+	"hakurei.app/internal/rosa"
+	"hakurei.app/message"
+	"hakurei.app/pkg"
+)
+
+// setSource is made available here to accept prepared hakurei tarballs.
+//
+//go:linkname setSource hakurei.app/internal/rosa.(*S).setSource
+func setSource(s *rosa.S, p []byte, version string)
+
+// inotifyInit returns a new inotify instance.
+func inotifyInit() (*os.File, error) {
+	fd, err := syscall.InotifyInit1(syscall.IN_NONBLOCK | syscall.IN_CLOEXEC)
+	if err != nil {
+		return nil, os.NewSyscallError("inotify_init1", err)
+	}
+	return os.NewFile(uintptr(fd), "inotify"), nil
+}
+
+// inotifyAddWatch adds pathname to in.
+func inotifyAddWatch(
+	in *os.File,
+	pathname string,
+	mask uint32,
+) (watchdesc int, err error) {
+	sc, _err := in.SyscallConn()
+	if _err != nil {
+		return -1, _err
+	}
+	if _err = sc.Control(func(fd uintptr) {
+		watchdesc, err = syscall.InotifyAddWatch(int(fd), pathname, mask)
+	}); _err != nil {
+		return -1, _err
+	}
+	return
+}
+
+// Follow reads from the file at pathname and writes its contents and any new
+// contents to w. follow returns if a read, write or inotify error occurs, or
+// the context is canceled.
+func Follow(ctx context.Context, pathname string, w io.Writer) (err error) {
+	var in *os.File
+	if in, err = inotifyInit(); err != nil {
+		return
+	}
+	defer func() {
+		if _err := in.Close(); err == nil {
+			err = _err
+		}
+	}()
+
+	if _, err = inotifyAddWatch(in, pathname, syscall.IN_MODIFY); err != nil {
+		return
+	}
+	var r *os.File
+	if r, err = os.Open(pathname); err != nil {
+		return
+	}
+	defer func() {
+		if _err := r.Close(); err == nil {
+			err = _err
+		}
+	}()
+
+	done := make(chan struct{})
+	defer close(done)
+	go func() {
+		select {
+		case <-ctx.Done():
+			now := time.Now()
+			_ = in.SetDeadline(now)
+			return
+
+		case <-done:
+			return
+		}
+	}()
+
+	if _, err = io.Copy(w, r); err != nil {
+		return
+	}
+
+	buf := make([]byte, os.Getpagesize())
+	for {
+		if _, err = io.Copy(w, r); err != nil {
+			return
+		}
+		if _, err = in.Read(buf); err != nil {
+			if errors.Is(err, os.ErrDeadlineExceeded) {
+				err = nil
+			}
+			return
+		}
+	}
+}
+
+// Path wraps the [pkg.Cache] pathname.
+type Path string
+
+// String returns the value of p.
+func (p Path) String() string { return string(p) }
+
+// append is [filepath.Join] with p as the first element.
+func (p Path) append(elem ...string) string {
+	return filepath.Join(append([]string{p.String()}, elem...)...)
+}
+
+// New returns a new [Path].
+func New(c *pkg.Cache) Path { return Path(c.Path().String()) }
+
+// name returns the CI socket pathname.
+func (p Path) name() string { return p.append("ci") }
+
+// client returns the CI http client.
+func (p Path) client() *http.Client {
+	var d net.Dialer
+	addr := net.UnixAddr{
+		Net:  "unix",
+		Name: p.name(),
+	}
+
+	return &http.Client{Transport: &http.Transport{
+		DialContext: func(ctx context.Context, _, _ string) (net.Conn, error) {
+			return d.DialUnix(ctx, "unix", nil, &addr)
+		},
+	}}
+}
+
+// cure writes the identifier of the pending artifact, cures the artifact, and
+// writes the cure whence. For an unsuccessful cure, a negative whence is
+// written, followed by a user-facing error string.
+func cure(c *pkg.Cache, a pkg.Artifact, w http.ResponseWriter) error {
+	h := w.Header()
+	h.Set("Content-Type", "application/octet-stream")
+	h.Set("Cache-Control", "no-cache")
+
+	f, ok := w.(http.Flusher)
+	if !ok {
+		_, _ = w.Write([]byte{0})
+		return errors.ErrUnsupported
+	}
+
+	id := c.Ident(a).Value()
+	if _, err := w.Write(id[:]); err != nil {
+		return err
+	}
+	f.Flush()
+
+	_, _, whence, err := c.CureWhence(a)
+	if err != nil {
+		whence = -1
+	}
+	if _, _err := w.Write(
+		binary.LittleEndian.AppendUint64(nil, uint64(whence)),
+	); _err != nil {
+		return _err
+	}
+
+	if err != nil {
+		if _, _err := io.WriteString(w, err.Error()); _err != nil {
+			return _err
+		}
+	}
+	f.Flush()
+	return err
+}
+
+// spool holds reusable [rosa.S] instances.
+var spool = sync.Pool{New: func() any { return rosa.New() }}
+
+// getS returns the address of a populated [rosa.S]. Its hakurei-source may be
+// clobbered and must be replaced using setSource before use.
+func getS() *rosa.S { return spool.Get().(*rosa.S) }
+
+// putS returns s to spool.
+func putS(s *rosa.S) { spool.Put(s) }
+
+// versionSize is the maximum size of the specified version string, plus its
+// deliminator byte.
+const versionSize = 8 + 16 + 6 + 2
+
+// errBadVersion is returned by readSource if a header does not contain
+// the deliminator byte.
+var errBadVersion = errors.New("unterminated version string")
+
+// readSource reads a version string and compressed source tarball from r and
+// returns the address of a [rosa.S] with this source tarball. The resulting
+// [rosa.S] must be returned via putS.
+func readSource(w http.ResponseWriter, r *http.Request) (*rosa.S, error) {
+	var header [versionSize]byte
+	_, err := io.ReadFull(r.Body, header[:])
+	if err != nil {
+		_ = r.Body.Close()
+		http.Error(w, "bad header", http.StatusBadRequest)
+		return nil, err
+	}
+
+	var version string
+	if i := bytes.IndexByte(header[:], 0); i < 0 {
+		_ = r.Body.Close()
+		http.Error(w, "unterminated version string", http.StatusBadRequest)
+		return nil, errBadVersion
+	} else {
+		version = unsafe.String(&header[0], i)
+	}
+
+	var p []byte
+	if p, err = io.ReadAll(r.Body); err != nil {
+		_ = r.Body.Close()
+		http.Error(w, "cannot receive payload", http.StatusInternalServerError)
+		return nil, err
+	}
+
+	s := getS()
+	setSource(s, p, version)
+	return s, r.Body.Close()
+}
+
+// ErrDaemonError is returned by writeSource generally if cure could not flush
+// on the connection to notify completion.
+var ErrDaemonError = errors.New("CI service could not process the request")
+
+// writeSource writes a source tarball to the specified endpoint of the CI
+// backend servicing the cache referred to by cm.
+func (p Path) writeSource(
+	ctx context.Context,
+	w io.Writer,
+	endpoint, source, version string,
+) (*pkg.ID, error) {
+	if len(version) >= versionSize {
+		return nil, syscall.ENOMEM
+	}
+	var header [versionSize]byte
+	copy(header[:], version[:])
+
+	var buf bytes.Buffer
+	gw := gzip.NewWriter(&buf)
+	tw := tar.NewWriter(gw)
+
+	if err := filepath.WalkDir(source, func(path string, d fs.DirEntry, err error) error {
+		if err != nil {
+			return err
+		}
+
+		if d.IsDir() && d.Name() == ".git" {
+			return fs.SkipDir
+		}
+
+		var fi fs.FileInfo
+		if fi, err = d.Info(); err != nil {
+			return err
+		}
+
+		var linkname string
+		if fi.Mode()&fs.ModeSymlink != 0 {
+			if linkname, err = os.Readlink(path); err != nil {
+				return err
+			}
+		}
+
+		var h *tar.Header
+		if h, err = tar.FileInfoHeader(fi, linkname); err != nil {
+			return err
+		}
+		h.Name = path
+
+		var isVersion bool
+		if dir, file := filepath.Split(path); filepath.Base(dir) == "dist" &&
+			file == "VERSION" && fi.Mode().IsRegular() {
+			isVersion = true
+			h.Size = int64(len(version))
+		}
+
+		if err = tw.WriteHeader(h); err != nil {
+			return err
+		}
+
+		if isVersion {
+			_, err = io.WriteString(tw, version)
+			return err
+		}
+
+		if fi.Mode().IsRegular() {
+			var f io.ReadCloser
+			if f, err = os.Open(path); err != nil {
+				return err
+			}
+
+			_, err = io.Copy(tw, f)
+			if _err := f.Close(); err == nil {
+				err = _err
+			}
+			if err != nil {
+				return err
+			}
+		}
+		return nil
+	}); err != nil {
+		return nil, err
+	}
+	if err := tw.Close(); err != nil {
+		return nil, err
+	}
+	if err := gw.Close(); err != nil {
+		return nil, err
+	}
+
+	req, err := http.NewRequestWithContext(
+		ctx,
+		http.MethodPost,
+		"http://"+path.Join("_", endpoint),
+		io.MultiReader(bytes.NewReader(header[:]), bytes.NewReader(buf.Bytes())),
+	)
+	if err != nil {
+		return nil, err
+	}
+
+	var resp *http.Response
+	resp, err = p.client().Do(req)
+	if err != nil {
+		return nil, err
+	}
+
+	var id pkg.ID
+	_, err = io.ReadFull(resp.Body, id[:])
+	if err != nil {
+		_ = resp.Body.Close()
+		if errors.Is(err, io.ErrUnexpectedEOF) {
+			return nil, ErrDaemonError
+		}
+		return nil, err
+	}
+
+	c, cancel := context.WithCancel(ctx)
+	done := make(chan error, 1)
+	var whence int
+	go func() {
+		defer cancel()
+		var wbuf [8]byte
+		_, _err := io.ReadFull(resp.Body, wbuf[:])
+		whence = int(binary.LittleEndian.Uint64(wbuf[:]))
+		done <- _err
+	}()
+
+	if w != nil {
+	retry:
+		err = Follow(c, filepath.Join(p.String(), "status", pkg.Encode(id)), w)
+		if err != nil {
+			if ctx.Err() == nil && errors.Is(err, os.ErrNotExist) {
+				goto retry
+			}
+
+			_ = resp.Body.Close()
+			return nil, err
+		}
+	}
+
+	err = <-done
+	if err != nil {
+		_ = resp.Body.Close()
+		return nil, err
+	}
+
+	if whence < 0 {
+		var m []byte
+		if m, err = io.ReadAll(resp.Body); err != nil {
+			_ = resp.Body.Close()
+			return nil, err
+		} else if err = resp.Body.Close(); err != nil {
+			return nil, err
+		}
+		return nil, errors.New(unsafe.String(unsafe.SliceData(m), len(m)))
+	}
+	return &id, resp.Body.Close()
+}
+
+// The stubKey is used by the mirror service exposed by serve where
+// authentication is unnecessary.
+var stubKey = ed25519.NewKeyFromSeed(make([]byte, ed25519.SeedSize))
+
+// fetch fetches the outcome of id and writes it to the specified directory.
+func (p Path) fetch(ctx context.Context, id *pkg.ID, output string) error {
+	c := p.client()
+	r, err := rosa.NewRemote(
+		c, "http://_",
+		stubKey.Public().(ed25519.PublicKey),
+	)
+	if err != nil {
+		return err
+	}
+
+	var sum *pkg.Checksum
+	if sum, err = r.Artifact(ctx, unique.Make(*id)); err != nil {
+		return err
+	}
+
+	var req *http.Request
+	if req, err = http.NewRequestWithContext(
+		ctx,
+		http.MethodGet,
+		"http://"+path.Join("_", "outcome", pkg.Encode(*sum)),
+		nil,
+	); err != nil {
+		return err
+	}
+
+	var resp *http.Response
+	if resp, err = c.Do(req); err != nil {
+		return err
+	}
+
+	err = pkg.Extract(resp.Body, output, nil)
+	if closeErr := resp.Body.Close(); err == nil {
+		err = closeErr
+	}
+	return err
+}
+
+// MakeDist creates a hakurei distribution using the CI service.
+func (p Path) MakeDist(
+	ctx context.Context,
+	w io.Writer,
+	output, source, version string,
+) error {
+	id, err := p.writeSource(ctx, w, "/dist", source, version)
+	if err != nil {
+		return err
+	}
+	return p.fetch(ctx, id, output)
+}
+
+// Serve services CI workload dispatched to c.
+func Serve(ctx context.Context, msg message.Msg, c *pkg.Cache) error {
+	const shutdownTimeout = 15 * time.Second
+	p := New(c)
+	addr := net.UnixAddr{
+		Net:  "unix",
+		Name: p.name(),
+	}
+
+	var mux http.ServeMux
+	mux.HandleFunc("POST /dist", func(w http.ResponseWriter, r *http.Request) {
+		s, err := readSource(w, r)
+		if err != nil {
+			msg.Verbose(err)
+			return
+		}
+		defer putS(s)
+
+		_, a := s.Std().MustLoad(rosa.H("hakurei-dist"))
+		if err = cure(c, a, w); err != nil {
+			msg.Verbose(err)
+			return
+		}
+		if msg.IsVerbose() {
+			msg.Verbosef(
+				"satisfied distribution %s",
+				pkg.Encode(c.Ident(a).Value()),
+			)
+		}
+	})
+
+	if r, err := os.OpenRoot(p.String()); err != nil {
+		return err
+	} else {
+		defer func() {
+			if err = r.Close(); err != nil {
+				msg.Verbose(err)
+			}
+		}()
+		rosa.NewMirror(msg, r.FS(), stubKey).Register(&mux)
+	}
+
+	server := http.Server{Handler: &mux}
+	go func() {
+		<-ctx.Done()
+		cc, cancel := context.WithTimeout(context.Background(), shutdownTimeout)
+		defer cancel()
+		if _err := server.Shutdown(cc); _err != nil {
+			log.Fatal(_err)
+		}
+	}()
+
+	ul, err := net.ListenUnix("unix", &addr)
+	if err != nil {
+		return err
+	}
+	ul.SetUnlinkOnClose(true)
+	msg.Verbosef("listening on %s", addr.Name)
+
+	err = server.Serve(ul)
+	if errors.Is(err, http.ErrServerClosed) {
+		err = nil
+	}
+	return err
+}