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"
	"net/http"
	"os"
	"os/signal"
	"slices"
	"strings"
	"sync"
	"syscall"
	"time"

	stdio "git.novox.be/novox/mesh-sdk/go"
	"git.novox.be/novox/mesh-sdk/go/objectlink"
	"git.novox.be/novox/mesh-sdk/go/process"
	"github.com/nats-io/nats.go/jetstream"

	"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, req ScanRequest) ([]Capture, error)
	// Convert calls the seat document-converter's convert verb as this process's own account (clause 7); nil: none.
	Convert func(body map[string]any) (json.RawMessage, error)
	// ConvertFollow is how long a conversion is followed; zero is ConvertFollow (a test makes it shorter).
	ConvertFollow time.Duration
	// Fetcher is the guarded fetch of a web link (fetch.go); it reads its facts when it is first needed.
	Fetcher func() (*Fetcher, 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
	// docs serialises the work on open documents.
	docs sync.Mutex
}

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

// tooLargeCode is the bus's error code for a message over the maximum allowed (the state refuses a value so).
const tooLargeCode = 10054

// 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; an answer the state refuses as too large is
// replaced by a small error answer and the message terminated; an answer that could not be written for another
// reason (the store away, a timeout) 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
	}
	if !ask.Deadline.IsZero() && w.Now().After(ask.Deadline) {
		// The asker has stopped waiting for this ask to be taken up: nothing is scanned or printed for it (issue
		// 1539). The reason is its answer, for a late reader, and the message ends here.
		w.logf("[brother] device: ask %s was taken up after its deadline %s; the device is not touched", ask.ID, ask.Deadline.Format(time.RFC3339))
		if _, err := w.Answers.Create(ask.ID, map[string]any{"error": "the ask was taken up after its deadline (the device process was busy): nothing was done for it; ask again"}); err != nil && !errors.Is(err, stdio.ErrChanged) {
			w.logf("[brother] device: the answer to %s could not be written: %v", ask.ID, err)
		}
		return actionTerm, 0
	}
	settled := false
	if (ask.Op == OpScan || ask.Op == OpPrint) && early != nil {
		early()
		settled = true
	}
	// A scan of the feeder may run as long as FeederMaxBound (the scan itself ends sooner by FeederBound, which
	// scales with its pages) and the PDF is then stored; every other ask is done in 5 minutes.
	bound := 5 * time.Minute
	if ask.Op == OpScan && ask.Source == "feeder" {
		bound = FeederMaxBound + 5*time.Minute
	}
	ctx, cancel := context.WithTimeout(context.Background(), bound)
	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)
		var api *jetstream.APIError
		if !errors.As(err, &api) || api.ErrorCode != tooLargeCode {
			// The store is away, timed out or cut: the answer may be written next time (a scan is settled already).
			if settled {
				return actionSettled, 0
			}
			return actionRetry, 5 * time.Second
		}
		// The ask ends here: a refused value would be refused 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, ask.Links, ask.Hidden)
	case OpScan:
		return w.scan(ctx, ask)
	case OpDocument:
		return w.document(ctx, ask)
	case OpUpload:
		return w.upload(ctx, ask)
	case OpPrint:
		return w.print(ctx, ask)
	case OpOpen:
		return w.openDocuments(ctx)
	}
	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 (cropped) or the feeder (whole sheets, one PDF of all pages), 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, ask Ask) (map[string]any, error) {
	req, err := ParseScan(ask.Source, ask.Duplex, ask.Apart)
	if err != nil {
		return nil, err
	}
	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}
	if ask.Document != "" {
		return w.scanDocument(ctx, ask, svc, objects)
	}
	capture := svc.Capture
	if w.Capture != nil {
		capture = w.Capture
	}
	caps, err := capture(ctx, req)
	if err != nil {
		return nil, err
	}
	return w.keep(ctx, objects, set, caps)
}

// keep stores each captured PDF in the bucket and says scan.added for it, in that order. One document is answered
// as today: the event's body with key, size and the number of parts. Several (documents lying apart on the glass)
// are answered as documents, a list of those, each filed on its own by the tool. A document that could not be kept
// or announced is named in failed and does not hold back the others: what is announced is always filed.
func (w *deviceWorker) keep(ctx context.Context, objects Objects, set Settings, caps []Capture) (map[string]any, error) {
	if len(caps) == 0 {
		return nil, fmt.Errorf("the scanner gave no document")
	}
	var docs []map[string]any
	var failed []string
	for i, c := range caps {
		one, err := w.keepOne(ctx, objects, set, c)
		if err != nil {
			if len(caps) == 1 {
				return nil, err
			}
			failed = append(failed, fmt.Sprintf("document %d: %v", i+1, err))
			continue
		}
		docs = append(docs, one)
	}
	if len(docs) == 0 {
		return nil, fmt.Errorf("no document could be kept: %s", strings.Join(failed, "; "))
	}
	if len(caps) == 1 {
		return docs[0], nil
	}
	out := map[string]any{"documents": docs}
	if len(failed) > 0 {
		out["failed"] = failed
	}
	return out, nil
}

// keepOne keeps one captured PDF and announces it.
func (w *deviceWorker) keepOne(ctx context.Context, objects Objects, set Settings, c Capture) (map[string]any, error) {
	// A scan over a document's limit is not kept nor announced: an announced scan is always filed (issue 1539).
	if len(c.PDF) > int(objectlink.MaxDocument) {
		return nil, fmt.Errorf("the scan's PDF is %d bytes, over the %d a document may be (novox/hq ADR 0468 §3): it is not kept; scan with a lower resolution or fewer sheets",
			len(c.PDF), objectlink.MaxDocument)
	}
	signer, ok := objects.(Presigner)
	if !ok {
		return nil, errors.New("the bucket cannot sign a link: the scan is not kept")
	}
	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)
	}
	// The answer carries a link to read the stored PDF, signed with the binding's key (novox/hq ADR 0468 §4): the tool
	// side reads the scan by it and files it; no byte of the document is in the answer.
	link, err := signer.Link(http.MethodGet, key, objectlink.MaxExpires)
	if err != nil {
		return nil, fmt.Errorf("the scan is kept as %s, but a link to it could not be signed: %w", key, err)
	}
	answer := map[string]any{"key": key, "size": len(c.PDF), "link": link, "link-seconds": int(objectlink.MaxExpires / time.Second)}
	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
	var doc []byte
	contentType := ""
	switch {
	case ask.URL != "":
		fetched, err := w.fetch(ctx, ask.URL)
		if err != nil {
			return nil, err
		}
		// The fetched document is kept once under its SHA-256, as a scan is, and printed as that object.
		doc, key, contentType = fetched.Body, sha256Hex(fetched.Body), fetched.ContentType
		if err := Store(ctx, objects, key, fetched.ContentType, doc); err != nil {
			return nil, err
		}
	case 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)
		}
		fallthrough
	default:
		var err error
		if doc, err = objects.Get(ctx, key); err != nil {
			return nil, err
		}
	}
	format, _, err := SniffFormat(doc)
	var convertedFrom string
	if err != nil {
		// Not a PDF or a JPEG: a type the converter takes becomes a PDF first (clause 7), any other is refused.
		pdf, pdfKey, cerr := w.toPDF(ctx, objects, key, doc, contentType)
		if cerr != nil {
			return nil, cerr
		}
		doc, convertedFrom, key = pdf, key, pdfKey
		format = "application/pdf"
	}
	// The printer is asked what it takes before the job, and a format it does not list is refused with the list
	// (novox/hq ADR 0468 clause 5): a printer that cannot be asked prints nothing.
	listed, err := FormatsOf(ctx, set.Printer)
	if err != nil {
		return nil, fmt.Errorf("the printer's document formats could not be read, so nothing is sent: %w", err)
	}
	if !slices.Contains(listed, format) {
		return nil, fmt.Errorf("the printer does not take %s: its document-format-supported lists %s", format, strings.Join(listed, ", "))
	}
	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
	}
	out := 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}
	if convertedFrom != "" {
		out["converted-from"] = convertedFrom
	}
	return out, nil
}

// fetch gets a web link under the rules of fetch.go.
func (w *deviceWorker) fetch(ctx context.Context, link string) (*Fetched, error) {
	if w.Fetcher == nil {
		return nil, errors.New("this process has no way to fetch a web link")
	}
	f, err := w.Fetcher()
	if err != nil {
		return nil, err
	}
	return f.Fetch(ctx, link)
}

// 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, links, hidden bool) (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)
	out := map[string]any{
		"page":   page,
		"title":  p.Title,
		"forms":  nonNilForms(withoutHidden(p.Forms, hidden)),
		"logout": logout,
	}
	if links {
		out["links"] = nonNil(p.Links)
	}
	return out, nil
}

// withoutHidden leaves the hidden fields (the CSRF token and the like, noise to a reader) out of the forms
// unless they are asked for.
func withoutHidden(forms []Form, keep bool) []Form {
	if keep {
		return forms
	}
	out := make([]Form, 0, len(forms))
	for _, f := range forms {
		var fields []Field
		for _, fl := range f.Fields {
			if fl.Type != "hidden" {
				fields = append(fields, fl)
			}
		}
		if len(fields) == 0 && len(f.Fields) > 0 {
			continue // a form of nothing but hidden fields (the logout form) has nothing to read
		}
		f.Fields = fields
		out = append(out, f)
	}
	return out
}

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, fetchFile := os.Getenv("BROTHER_STORE"), os.Getenv("BROTHER_FETCH")
	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
		},
		Fetcher: func() (*Fetcher, error) {
			r, err := ReadFetchFacts(fetchFile)
			if err != nil {
				return nil, err
			}
			return NewFetcher(r), 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)
	w.Convert = func(body map[string]any) (json.RawMessage, error) {
		return conn.Call("seat:document-converter.convert", body)
	}
	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)
		}
	})
}

// An open document (issue 1538) is kept in the bucket, not in memory, so that a restart of the device process
// loses none of its pages: each page's JPEG under pending/<id>/<n>.jpg and a manifest, pending/<id>/manifest.json,
// that says how many pages there are and, once it is closed, the key of its PDF. The pages are not deleted when it
// is closed (the bucket has no delete); they are the bucket's to age out.
const mostDocumentPages = 100

type docPage struct {
	Key     string `json:"key"`
	DPI     int    `json:"dpi"`
	Width   int    `json:"width"`
	Height  int    `json:"height"`
	Cropped bool   `json:"cropped"`
}

type docManifest struct {
	Started  time.Time `json:"started"`
	Updated  time.Time `json:"updated"` // the last page's time
	Pages    []docPage `json:"pages"`
	Finished string    `json:"finished,omitempty"` // the closed document's object key
}

func manifestKey(id string) string { return "pending/" + id + "/manifest.json" }

func loadManifest(ctx context.Context, o Objects, id string) (*docManifest, error) {
	has, err := o.Has(ctx, manifestKey(id))
	if err != nil {
		return nil, err
	}
	if !has {
		return nil, fmt.Errorf("no document %s is open: its id is wrong, or it was closed", id)
	}
	raw, err := o.Get(ctx, manifestKey(id))
	if err != nil {
		return nil, err
	}
	var m docManifest
	if err := json.Unmarshal(raw, &m); err != nil {
		return nil, fmt.Errorf("document %s: its manifest cannot be read: %w", id, err)
	}
	if m.Finished != "" {
		return nil, fmt.Errorf("document %s is closed already, as %s", id, m.Finished)
	}
	return &m, nil
}

func saveManifest(ctx context.Context, o Objects, id string, m *docManifest) error {
	raw, _ := json.Marshal(m)
	return o.Put(ctx, manifestKey(id), "application/json", raw)
}

// scanDocument adds a glass page to an open document (document start opens one, with the ask's id as its id), or
// with finish closes the document into one multi-page PDF, kept and announced once like any scan.
func (w *deviceWorker) scanDocument(ctx context.Context, ask Ask, svc *Service, objects Objects) (map[string]any, error) {
	if ask.Source == "feeder" || ask.Duplex {
		return nil, fmt.Errorf("a document of several pages is scanned from the glass, page by page; the feeder takes the whole stack in one scan")
	}
	id := ask.Document
	if id == "start" {
		if ask.Finish {
			return nil, fmt.Errorf("finish closes an open document: name its id, not start")
		}
		id = ask.ID
	}
	if !uploadPattern.MatchString(id) {
		return nil, fmt.Errorf("document %q is not an id: letters, digits, - and _, at most 64", id)
	}
	w.docs.Lock()
	defer w.docs.Unlock()
	var m *docManifest
	if ask.Document == "start" {
		m = &docManifest{Started: w.Now().UTC(), Updated: w.Now().UTC()}
	} else {
		var err error
		if m, err = loadManifest(ctx, objects, id); err != nil {
			return nil, err
		}
	}
	if ask.Finish {
		return w.finishDocument(ctx, id, m, svc, objects)
	}
	if len(m.Pages) >= mostDocumentPages {
		return nil, fmt.Errorf("document %s has %d pages, the most one takes: close it", id, len(m.Pages))
	}
	page, err := svc.GlassPage(ctx)
	if err != nil {
		return nil, err
	}
	key := fmt.Sprintf("pending/%s/%03d.jpg", id, len(m.Pages)+1)
	if err := objects.Put(ctx, key, "image/jpeg", page.JPEG); err != nil {
		return nil, err
	}
	m.Updated = w.Now().UTC()
	m.Pages = append(m.Pages, docPage{Key: key, DPI: page.DPI, Width: page.Width, Height: page.Height, Cropped: page.Cropped})
	if err := saveManifest(ctx, objects, id, m); err != nil {
		return nil, err
	}
	return map[string]any{"document": id, "pages": len(m.Pages), "state": "open",
		"next": "scanner.scan {document: \"" + id + "\"} adds a page; scanner.scan {document: \"" + id + "\", finish: true} closes it into one PDF"}, nil
}

// finishDocument builds the open document's pages into one PDF and keeps and announces it.
func (w *deviceWorker) finishDocument(ctx context.Context, id string, m *docManifest, svc *Service, objects Objects) (map[string]any, error) {
	if len(m.Pages) == 0 {
		return nil, fmt.Errorf("document %s has no page", id)
	}
	pages := make([]PDFPage, len(m.Pages))
	for i, p := range m.Pages {
		jpg, err := objects.Get(ctx, p.Key)
		if err != nil {
			return nil, fmt.Errorf("document %s: page %d is not kept: %w", id, i+1, err)
		}
		pages[i] = PDFPage{JPEG: jpg, DPI: p.DPI}
	}
	first := m.Pages[0]
	c, err := captureOf(pages, Cropped{Width: first.Width, Height: first.Height, Cropped: first.Cropped}, first.DPI, w.Now())
	if err != nil {
		return nil, err
	}
	set, err := w.Settings()
	if err != nil {
		return nil, err
	}
	answer, err := w.keepOne(ctx, objects, set, c)
	if err != nil {
		return nil, err
	}
	m.Finished = answer["key"].(string)
	if err := saveManifest(ctx, objects, id, m); err != nil {
		return nil, fmt.Errorf("document %s is kept as %s, but could not be marked closed: %w", id, m.Finished, err)
	}
	// The pages are not needed any more: delete them. A bucket that refuses leaves them, and the answer says so.
	for _, p := range m.Pages {
		if err := objects.Delete(ctx, p.Key); err != nil {
			answer["pending-kept"] = fmt.Sprintf("the pages of document %s could not be deleted from the bucket and stay under pending/%s/: %v", id, id, err)
			break
		}
	}
	return answer, nil
}

// openDocuments lists the documents of several pages that are open in the bucket, with the time of each one's last
// page, so that the tool can arm their expiry again after it was restarted.
func (w *deviceWorker) openDocuments(ctx context.Context) (map[string]any, error) {
	objects, err := w.Objects()
	if err != nil {
		return nil, err
	}
	keys, err := objects.List(ctx, "pending/")
	if err != nil {
		return nil, err
	}
	open := []map[string]any{}
	for _, k := range keys {
		rest, ok := strings.CutPrefix(k, "pending/")
		id, name, _ := strings.Cut(rest, "/")
		if !ok || name != "manifest.json" || !uploadPattern.MatchString(id) {
			continue
		}
		m, err := loadManifest(ctx, objects, id)
		if err != nil {
			continue // closed, or unreadable: not open
		}
		at := m.Updated
		if at.IsZero() {
			at = m.Started
		}
		open = append(open, map[string]any{"id": id, "pages": len(m.Pages), "updated": at.UTC().Format(time.RFC3339)})
	}
	return map[string]any{"open": open}, nil
}