mesh-catalog/modules/brother/cmd/brother/deviceproc.go

package main

// mesh brother device (novox/hq ADR 0467 clause 4, in the shape of ADR 0436 §7): the device process, a process
// with a bus account of its own (brother-device), never a seat. It alone of the module's parts reads the admin
// password and logs in to the printer's admin page; the tools queue an ask on the work-queue stream device-asks
// and read its answer in the state device-answers. The admin password is never in an answer, an event or a log.

import (
	"context"
	"encoding/base64"
	"encoding/json"
	"errors"
	"fmt"
	"os"
	"os/signal"
	"sync"
	"syscall"
	"time"

	stdio "git.novox.be/novox/mesh-sdk/go"
	"git.novox.be/novox/mesh-sdk/go/process"

	"brother/cmd/brother/internal/secret"
)

// Answers is the state the worker writes an answer in, once.
type Answers interface {
	Create(key string, value any) (uint64, error)
}

// action is what is done to the queued message.
type action int

const (
	actionAck action = iota
	actionRetry
	actionTerm
	// actionSettled says the message was acknowledged already, before the work began: what is not safe to do twice
	// (a scan, a print) is acknowledged first, so a slow one is not delivered again after the bus's wait.
	actionSettled
)

// pendingUpload is the parts of an upload that have come.
type pendingUpload struct {
	parts map[int][]byte
	n     int
	sha   string
	at    time.Time
}

// pendingFor is how long an incomplete upload's parts are kept.
const pendingFor = 15 * time.Minute

// deviceWorker answers the asks: it alone talks to the device (the scanner, the printer's pages and IPP) and to
// the bucket. Settings and the password are read at each ask, so a changed config value or a password given later
// is the next ask's.
type deviceWorker struct {
	Settings func() (Settings, error)
	Password func() (string, error)
	Objects  func() (Objects, error)
	// Emit says scan.added on the bus under a message id.
	Emit    func(body map[string]any, id string) error
	Answers Answers
	Node    string
	// Capture scans; nil is the device's own (a test puts a stand-in).
	Capture func(ctx context.Context) (Capture, error)
	// answerPart is the raw size of a document part in an answer; zero is AnswerPartSize (a test makes it larger).
	answerPart int
	Now        func() time.Time
	Logf       func(format string, a ...any)

	mu      sync.Mutex
	pending map[string]*pendingUpload
}

func (w *deviceWorker) logf(format string, a ...any) {
	if w.Logf != nil {
		w.Logf(format, a...)
	}
}

// Handle answers one queued ask. An ask that cannot be read is terminated; an ask too old for anyone to wait for
// is acknowledged and the device not touched; every other outcome, a failure included, is written as the answer
// under the ask's id and the message acknowledged; only an answer that could not be written is redelivered (the
// bus's limit ends that). early acknowledges the message; it is called before a scan or a print, which are not
// done twice.
func (w *deviceWorker) Handle(body []byte, early func()) (action, time.Duration) {
	var ask Ask
	if err := json.Unmarshal(body, &ask); err != nil || ask.ID == "" {
		w.logf("[brother] device: a queued ask could not be read: %v", err)
		return actionTerm, 0
	}
	if age := w.Now().Sub(ask.At); age > StaleAfter {
		w.logf("[brother] device: ask %s is %s old; nobody waits for it, the device is not touched", ask.ID, age.Round(time.Second))
		return actionAck, 0
	}
	settled := false
	if (ask.Op == OpScan || ask.Op == OpPrint) && early != nil {
		early()
		settled = true
	}
	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Minute)
	defer cancel()
	var answer map[string]any
	if got, err := w.run(ctx, ask); err != nil {
		answer = map[string]any{"error": err.Error()}
	} else {
		answer = got
	}
	if _, err := w.Answers.Create(ask.ID, answer); err != nil && !errors.Is(err, stdio.ErrChanged) {
		w.logf("[brother] device: the answer to %s could not be written: %v", ask.ID, err)
		// The ask ends here: a refused answer would be refused again, and a redelivery would do the work again.
		// Say the failure as a small error answer, so the tool's wait ends with the reason (issue 1535).
		msg := fmt.Sprintf("the answer to this ask could not be written in the state %s: %v", AnswersState, err)
		if _, err2 := w.Answers.Create(ask.ID, map[string]any{"error": msg}); err2 != nil && !errors.Is(err2, stdio.ErrChanged) {
			w.logf("[brother] device: the failure of %s could not be written either: %v", ask.ID, err2)
		}
		if settled {
			return actionSettled, 0
		}
		return actionTerm, 0
	}
	if settled {
		return actionSettled, 0
	}
	return actionAck, 0
}

// run is the operation an ask names.
func (w *deviceWorker) run(ctx context.Context, ask Ask) (map[string]any, error) {
	switch ask.Op {
	case OpSettings:
		return w.read(ctx, ask.Page)
	case OpScan:
		return w.scan(ctx)
	case OpDocument:
		return w.document(ctx, ask)
	case OpUpload:
		return w.upload(ctx, ask)
	case OpPrint:
		return w.print(ctx, ask)
	}
	return nil, fmt.Errorf("the ask %s names an operation, %q, the device process does not do", ask.ID, ask.Op)
}

// scan scans the glass, crops, builds the PDF, keeps it in the bucket under its SHA-256 and says scan.added with
// the object's key (ADR 0468 clause 2), in that order: nothing is said of a scan that is not kept, and nothing is
// kept nor said when the scanner gives no page (1440/AC6). The answer is the event's body with key, size and the
// number of parts.
func (w *deviceWorker) scan(ctx context.Context) (map[string]any, error) {
	set, err := w.Settings()
	if err != nil {
		return nil, err
	}
	if set.Scanner == "" {
		return nil, unsetError("scanner")
	}
	objects, err := w.Objects()
	if err != nil {
		return nil, err
	}
	svc := &Service{Settings: set, Node: w.Node, Now: w.Now}
	capture := svc.Capture
	if w.Capture != nil {
		capture = w.Capture
	}
	c, err := capture(ctx)
	if err != nil {
		return nil, err
	}
	body := c.Body(w.Node, set.ColorMode)
	key := body["object"].(string)
	if err := Store(ctx, objects, key, "application/pdf", c.PDF); err != nil {
		return nil, err
	}
	if err := w.Emit(body, c.SHA256); err != nil {
		return nil, fmt.Errorf("the scan is kept as %s, but %s could not be said on the bus: %w", key, EventAdded, err)
	}
	answer := map[string]any{"key": key, "size": len(c.PDF), "parts": answerPartCount(len(c.PDF))}
	for k, v := range body {
		answer[k] = v
	}
	return answer, nil
}

// document is one part of a scan kept in the bucket.
func (w *deviceWorker) document(ctx context.Context, ask Ask) (map[string]any, error) {
	if err := checkDocument(ask); err != nil {
		return nil, err
	}
	objects, err := w.Objects()
	if err != nil {
		return nil, err
	}
	data, err := objects.Get(ctx, ask.Object)
	if err != nil {
		return nil, err
	}
	each := w.answerPart
	if each <= 0 {
		each = AnswerPartSize
	}
	part, n, err := cutPartOf(data, ask.Part, each)
	if err != nil {
		return nil, err
	}
	return map[string]any{"part": ask.Part, "parts": n, "size": len(data), "sha256": sha256Hex(data),
		"data": base64.StdEncoding.EncodeToString(part)}, nil
}

// upload keeps a part, and once all have come joins them, checks the SHA-256 and the format and stores the document
// as uploads/<upload>.<ext>. A wrong SHA-256 or a document that is neither PDF nor JPEG stores nothing, and the
// parts are dropped; a part out of range is refused.
func (w *deviceWorker) upload(ctx context.Context, ask Ask) (map[string]any, error) {
	if err := checkUpload(ask); err != nil {
		return nil, err
	}
	raw, _ := base64.StdEncoding.DecodeString(ask.Data)
	w.mu.Lock()
	if w.pending == nil {
		w.pending = map[string]*pendingUpload{}
	}
	for id, p := range w.pending {
		if w.Now().Sub(p.at) > pendingFor {
			delete(w.pending, id)
		}
	}
	p := w.pending[ask.Upload]
	if p == nil {
		p = &pendingUpload{parts: map[int][]byte{}, n: ask.Parts, sha: ask.SHA256}
		w.pending[ask.Upload] = p
	}
	if p.n != ask.Parts || p.sha != ask.SHA256 {
		w.mu.Unlock()
		return nil, fmt.Errorf("upload %s was begun with %d parts and SHA-256 %s: its parts all say the same", ask.Upload, p.n, p.sha)
	}
	p.parts[ask.Part] = raw
	p.at = w.Now()
	if len(p.parts) < p.n {
		got := len(p.parts)
		w.mu.Unlock()
		return map[string]any{"upload": ask.Upload, "complete": false, "received": got, "parts": ask.Parts}, nil
	}
	delete(w.pending, ask.Upload)
	w.mu.Unlock()

	ordered := make([][]byte, p.n)
	for i := range ordered {
		ordered[i] = p.parts[i+1]
	}
	doc, err := joinParts(ordered, p.sha)
	if err != nil {
		return nil, fmt.Errorf("upload %s: %w", ask.Upload, err)
	}
	format, ext, err := SniffFormat(doc)
	if err != nil {
		return nil, fmt.Errorf("upload %s: %w", ask.Upload, err)
	}
	objects, err := w.Objects()
	if err != nil {
		return nil, err
	}
	key := "uploads/" + ask.Upload + "." + ext
	if err := objects.Put(ctx, key, format, doc); err != nil {
		return nil, err
	}
	return map[string]any{"upload": ask.Upload, "complete": true, "object": key, "size": len(doc), "sha256": p.sha}, nil
}

// print fetches a scan or a completed upload from the bucket and sends it to the printer as an IPP Print-Job in
// its own format, and answers the job's id and state. The printer's host is its IPP address; nothing is fetched
// or sent before the ask is checked.
func (w *deviceWorker) print(ctx context.Context, ask Ask) (map[string]any, error) {
	if err := checkPrint(&ask); err != nil {
		return nil, err
	}
	set, err := w.Settings()
	if err != nil {
		return nil, err
	}
	if set.Printer == "" {
		return nil, unsetError("printer")
	}
	objects, err := w.Objects()
	if err != nil {
		return nil, err
	}
	key := ask.Object
	if key == "" {
		for _, ext := range []string{"pdf", "jpg"} {
			k := "uploads/" + ask.Upload + "." + ext
			if has, err := objects.Has(ctx, k); err != nil {
				return nil, err
			} else if has {
				key = k
				break
			}
		}
		if key == "" {
			return nil, fmt.Errorf("upload %s is not complete: no document is kept under it", ask.Upload)
		}
	}
	doc, err := objects.Get(ctx, key)
	if err != nil {
		return nil, err
	}
	format, _, err := SniffFormat(doc)
	if err != nil {
		return nil, err
	}
	res, err := PrintJob(ctx, set.Printer, PrintOptions{Format: format, Copies: ask.Copies, Sides: ask.Sides, Media: ask.Media, Name: "brother print"}, doc)
	if err != nil {
		return nil, err
	}
	return map[string]any{"job-id": res.JobID, "job-state": res.State, "job-state-code": res.StateCode,
		"job-state-reasons": res.Reasons, "format": format, "copies": ask.Copies, "sides": ask.Sides, "media": ask.Media, "object": key}, nil
}

// read logs in, reads one page of the admin site and logs out. It is read-only: the login (and the logout) are
// the only things it sends, and a password field's value is never in the answer. The printer is not contacted
// before the settings and the password are known to be given (1453/AC4).
func (w *deviceWorker) read(ctx context.Context, page string) (map[string]any, error) {
	if page == "" {
		page = DefaultPage
	}
	if !validPage(page) {
		return nil, fmt.Errorf("page %q is not a page of the admin site: it starts with / and ends in .html", page)
	}
	set, err := w.Settings()
	if err != nil {
		return nil, err
	}
	if err := set.MissingPrinter(); err != nil {
		return nil, err
	}
	pw, err := w.Password()
	if err != nil {
		return nil, err
	}
	a, err := NewAdmin(set.Printer, set.PrinterCertificate, pw)
	if err != nil {
		return nil, err
	}
	if err := a.Login(ctx); err != nil {
		return nil, err
	}
	// Log out with a context of its own: the ask's may be what ended this.
	lctx, cancel := context.WithTimeout(context.Background(), 15*time.Second)
	defer cancel()
	body, at, err := a.Get(ctx, page)
	if err != nil {
		if home, hat, herr := a.Get(lctx, DefaultPage); herr == nil {
			a.Logout(lctx, home, hat)
		}
		var nf *notFoundError
		if errors.As(err, &nf) {
			return nil, fmt.Errorf("the printer has no page %s", page)
		}
		return nil, fmt.Errorf("page %s cannot be read: %w", page, err)
	}
	logout := a.Logout(lctx, body, at)
	p := ParsePage(body)
	return map[string]any{
		"page":   page,
		"title":  p.Title,
		"links":  nonNil(p.Links),
		"forms":  nonNilForms(p.Forms),
		"logout": logout,
	}, nil
}

func nonNil(l []Link) []Link {
	if l == nil {
		return []Link{}
	}
	return l
}

func nonNilForms(f []Form) []Form {
	if f == nil {
		return []Form{}
	}
	return f
}

// device is `brother device`: it reads its own bus account, connects, and consumes device-asks until stopped.
// It refuses to start naming what is missing when the account cannot be read, and never logs a password.
func device() int {
	settings, passwordFile, brokerFile := os.Getenv("BROTHER_SETTINGS"), os.Getenv("BROTHER_ADMIN_PASSWORD"), os.Getenv("BROTHER_BROKER")
	store := os.Getenv("BROTHER_STORE")
	if brokerFile == "" {
		fmt.Fprintln(os.Stderr, "[brother] device was started without BROTHER_BROKER, its bus account's file")
		return 2
	}
	account, err := process.ReadAccount(brokerFile)
	if err != nil {
		fmt.Fprintf(os.Stderr, "[brother] device: %v\n", err)
		return 1
	}
	ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM)
	defer stop()
	w := &deviceWorker{
		Settings: func() (Settings, error) { return LoadSettings(settings) },
		Password: func() (string, error) { return secret.Read(passwordFile) },
		Objects: func() (Objects, error) {
			facts, err := ReadStoreFacts(store)
			if err != nil {
				return nil, err
			}
			return NewS3(facts), nil
		},
		Now:  time.Now,
		Logf: func(f string, a ...any) { fmt.Fprintf(os.Stderr, f+"\n", a...) },
	}
	fmt.Fprintf(os.Stderr, "[brother] device is consuming %s as %s\n", AsksStream, account.User)
	err = runDevice(ctx, account, w)
	if err != nil && ctx.Err() == nil {
		fmt.Fprintf(os.Stderr, "[brother] device stopped: %v\n", err)
		return 1
	}
	return 0
}

// runDevice connects as the device process's own account and consumes device-asks until ctx ends, answering each
// ask in the state device-answers on that connection and then settling the message as Handle says. Apart from
// device so a test can drive it over a bus of its own.
func runDevice(ctx context.Context, account process.Account, w *deviceWorker) error {
	conn, err := process.Connect(account)
	if err != nil {
		return fmt.Errorf("the bus cannot be reached: %w", err)
	}
	defer conn.Close()
	w.Answers = conn.State(AnswersState)
	if w.Node == "" {
		w.Node = account.Node
	}
	w.Emit = func(body map[string]any, id string) error {
		return conn.Emit(EventAdded, body, stdio.EmitOptions{MessageID: id})
	}
	return conn.Consume(ctx, AsksStream, func(msg *process.Message) {
		var settleErr error
		act, delay := w.Handle(msg.Body, func() { settleErr = msg.Ack() })
		if settleErr != nil {
			w.logf("[brother] device: acknowledging %s early: %v", msg.Subject, settleErr)
		}
		var aerr error
		switch act {
		case actionAck:
			aerr = msg.Ack()
		case actionRetry:
			aerr = msg.NakWithDelay(delay)
		case actionTerm:
			aerr = msg.Term()
		}
		if aerr != nil {
			w.logf("[brother] device: settling %s: %v", msg.Subject, aerr)
		}
	})
}