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

package main

import (
	"bytes"
	"context"
	"fmt"
	"net/http"
	"strings"
	"sync"
	"time"

	"git.novox.be/novox/mesh-sdk/go/objectlink"
)

// EventAdded is the module's event: a scan was kept in the bucket (ADR 0464, ADR 0468 clause 2), said by the device
// process. Consumers declare brother.scan.added; it
// travels as mesh.mod.brother.event.scan.added. It names the object and never carries the image.
const EventAdded = "scan.added"

// Service is what the tools do, over one reading of the settings. The device process's own scan is Capture's.
type Service struct {
	Settings Settings
	// Node is this machine's name, for the event.
	Node string
	Now  func() time.Time
	// Asks is how the verbs reach the device process, which alone holds a credential.
	Asks AskBus
	// AnswerWait is how long a verb waits for the device process's answer (ScanWait for a scan); AnswerPoll how
	// often it looks.
	AnswerWait, ScanWait, AnswerPoll time.Duration
	// QuickWait is how long a feeder scan is waited for in one call, FeederWait how long it is followed.
	QuickWait, FeederWait time.Duration
	// Jobs are the scans this process follows.
	Jobs *Jobs
	// Upload asks a machine's node-files to send one of its files into the bucket and answers the key; Bucket is
	// brother's bucket. Both are the tool side's (printer.print with path and node).
	// Fetch reads the links the device process signs; DocExpiry and DocRetry are the open documents' waits. They are
	// copied into the Service when it is made, because background work reads them while a test (or a reload) changes the
	// package's own, which would race.
	Fetch               *http.Client
	DocExpiry, DocRetry time.Duration
	Upload              func(node, path, bucket string) (string, error)
	Bucket              func() (string, error)
}

// Jobs are the scans this tool process has started and not yet been asked the outcome of. A scan runs in the
// background from the moment it is asked for, so that it is filed whether or not the caller still waits (issue
// 1539); the caller waits for it, and reads its outcome later with its id when it did not finish in time.
type Jobs struct {
	mu        sync.Mutex
	m         map[string]*job
	open      map[string]*time.Timer // open documents: the timer that closes each
	recovered bool                   // the open documents of the bucket were looked for
}

// arm (re)starts the timer that closes the open document after d.
func (j *Jobs) arm(doc string, d time.Duration, closeIt func()) {
	j.mu.Lock()
	defer j.mu.Unlock()
	if j.open == nil {
		j.open = map[string]*time.Timer{}
	}
	if t := j.open[doc]; t != nil {
		t.Stop()
	}
	j.open[doc] = time.AfterFunc(d, func() {
		j.disarm(doc)
		closeIt()
	})
}

// disarm forgets the open document's timer.
func (j *Jobs) disarm(doc string) {
	j.mu.Lock()
	defer j.mu.Unlock()
	if t := j.open[doc]; t != nil {
		t.Stop()
		delete(j.open, doc)
	}
}

type job struct {
	at   time.Time
	done chan struct{}
	body map[string]any // set when done
	err  error          // set when failed
}

// jobKeep is how long a finished job's outcome can be read with its id.
var jobKeep = time.Hour

// docRetry is how long after a failed closing of an expired document it is tried again.
var docRetry = 5 * time.Minute

// NewJobs is an empty set of jobs.
func NewJobs() *Jobs { return &Jobs{m: map[string]*job{}} }

func (j *Jobs) get(id string) *job {
	if j == nil {
		return nil
	}
	j.mu.Lock()
	defer j.mu.Unlock()
	return j.m[id]
}

func (j *Jobs) put(id string) *job {
	j.mu.Lock()
	defer j.mu.Unlock()
	if j.m == nil {
		j.m = map[string]*job{}
	}
	// A job nobody has asked after for jobKeep is forgotten (a finished one only).
	for k, o := range j.m {
		select {
		case <-o.done:
			if time.Since(o.at) > jobKeep {
				delete(j.m, k)
			}
		default:
		}
	}
	x := &job{at: time.Now(), done: make(chan struct{})}
	j.m[id] = x
	return x
}

// ScanNow asks the device process to scan, crop, build the PDF, keep it in the bucket and say scan.added, waits for
// the stored scan's key, fetches its parts through the device process, joins them, checks their SHA-256 and files
// the PDF in the scan folder (temp and rename, no overwrite). It answers the event's body and the path. Nothing is
// asked before scanner and scan-folder are known to be set (1440/AC7); the tool never reads a credential.
//
// The scan is followed by a goroutine of its own to the end, so it is filed even when the caller has gone. A glass
// scan is waited for (ScanWait); a feeder scan only for QuickWait, and when it is not finished by then the answer is
// started: true with the scan's id, and ScanResult reads the outcome later.
func (s *Service) ScanNow(ctx context.Context, source string, duplex, apart bool) (map[string]any, error) {
	if err := s.Settings.Missing(); err != nil {
		return nil, err
	}
	req, err := ParseScan(source, duplex, apart)
	if err != nil {
		return nil, err
	}
	run, quick := scanRun, s.ScanWait
	if req.Feeder {
		run, quick = feederRun, s.QuickWait
	}
	return s.runScan(ctx, Ask{Op: OpScan, Source: source, Duplex: duplex, Apart: apart}, run, quick)
}

// runScan queues the ask and follows it in the background to its end (filing included); the caller is given the
// outcome if it comes within quick, and started: true with the id otherwise.
func (s *Service) runScan(ctx context.Context, ask Ask, run, quick time.Duration) (map[string]any, error) {
	id, j, err := s.startScan(ask, run)
	if err != nil {
		return nil, err
	}
	select {
	case <-j.done:
		return j.body, j.err
	case <-time.After(quick):
		return map[string]any{"started": true, "id": id, "state": "running",
			"note": "the scan goes on and is filed when it is done; read its outcome with scanner.scan {id}, or wait for scan.added"}, nil
	case <-ctx.Done():
		return nil, fmt.Errorf("the caller went away; the scan %s goes on and is filed: %w", id, ctx.Err())
	}
}

// startScan queues the ask and starts the goroutine that follows it, however long the caller waits.
//
// The ask carries a deadline, ScanWait from now: the device process drops it unscanned if it is taken up later (it
// may be busy with a long feeder scan). One taken up in time may run for run, and the goroutine waits for that
// too, so a scan that is announced is always filed, however long the caller has waited.
func (s *Service) startScan(ask Ask, run time.Duration) (string, *job, error) {
	startBy := s.ScanWait
	bound := startBy + run
	id, err := s.queue(ask, startBy)
	if err != nil {
		return "", nil, err
	}
	j := s.Jobs.put(id)
	go func() {
		// The goroutine outlives the caller: a context of its own, bounded by the scan's own bound and the time
		// to fetch the parts.
		ctx, cancel := context.WithTimeout(context.Background(), bound+10*time.Minute)
		defer cancel()
		j.body, j.err = s.finishScan(ctx, id, bound, ask)
		close(j.done)
	}()
	return id, j, nil
}

// docExpiry is how long an open document waits for its next page before its pages are filed as one PDF anyway.
var docExpiry = 30 * time.Minute

// ScanPage scans the glass into a document of several pages (issue 1538). document is start, which opens a
// document and scans its first page into it, or the id of an open one, which scans the next page into it; with
// finish the document is closed into one PDF, filed once (nothing is scanned). An open document that gets no page
// for docExpiry is closed and filed so, by this process: its pages are never lost.
func (s *Service) ScanPage(ctx context.Context, document string, finish bool) (map[string]any, error) {
	if err := s.Settings.Missing(); err != nil {
		return nil, err
	}
	if document != "start" && !uploadPattern.MatchString(document) {
		return nil, fmt.Errorf("document is start, or the id of an open document: letters, digits, - and _, at most 64")
	}
	if finish && document == "start" {
		return nil, fmt.Errorf("finish closes an open document: name its id in document, not start")
	}
	return s.runScan(ctx, Ask{Op: OpScan, Document: document, Finish: finish}, scanRun, s.ScanWait)
}

// ScanResult is the outcome of the scan with this id: its body and path when it is done, state running when it is
// not, and state failed with why when it failed. A scan this process does not know (it was restarted, or the id is
// wrong) is an error.
func (s *Service) ScanResult(id string) (map[string]any, error) {
	j := s.Jobs.get(id)
	if j == nil {
		return nil, fmt.Errorf("no scan %s is known to this tool: its id is wrong, or the tool was restarted since it was asked for (a scan the device process kept is announced by scan.added)", id)
	}
	select {
	case <-j.done:
		if j.err != nil {
			return map[string]any{"id": id, "state": "failed", "why": j.err.Error()}, nil
		}
		out := map[string]any{"id": id, "state": "done"}
		for k, v := range j.body {
			out[k] = v
		}
		return out, nil
	default:
		return map[string]any{"started": true, "id": id, "state": "running"}, nil
	}
}

// finishScan waits for the device process's answer to the scan ask, then fetches the stored scan and files it.
func (s *Service) finishScan(ctx context.Context, id string, wait time.Duration, ask Ask) (map[string]any, error) {
	ans, err := s.await(ctx, id, wait)
	if err != nil {
		return nil, err
	}
	if doc, _ := ans["document"].(string); doc != "" && ans["key"] == nil {
		// A page was added to an open document: nothing to file yet, and the document is closed if it is left open.
		s.armDocument(doc, s.docExpiry(), 0)
		return ans, nil
	}
	if ask.Document != "" && ask.Finish {
		out, err := s.fileAnswer(ctx, ans)
		if err == nil {
			s.Jobs.disarm(ask.Document)
		}
		return out, err
	}
	return s.fileAnswer(ctx, ans)
}

// fileAnswer fetches the scan the device process's answer names and files it; an answer that lists documents (several
// lying apart on the glass, issue 1528) is each fetched and filed as a PDF of its own, and answered as documents,
// in the order the scanner found them. One that cannot be filed does not hold back the others: it is named in failed.
func (s *Service) fileAnswer(ctx context.Context, ans map[string]any) (map[string]any, error) {
	list, many := ans["documents"].([]any)
	if !many {
		return s.fileOne(ctx, ans)
	}
	var filed []any
	var failed []string
	for i, d := range list {
		one, _ := d.(map[string]any)
		body, err := s.fileOne(ctx, one)
		if err != nil {
			failed = append(failed, fmt.Sprintf("document %d: %v", i+1, err))
			continue
		}
		filed = append(filed, body)
	}
	if bad, ok := ans["failed"].([]any); ok {
		for _, b := range bad {
			failed = append(failed, fmt.Sprint(b))
		}
	}
	if len(filed) == 0 {
		return nil, fmt.Errorf("no document could be filed: %s", strings.Join(failed, "; "))
	}
	out := map[string]any{"documents": filed, "count": len(filed)}
	if len(failed) > 0 {
		out["failed"] = failed
	}
	return out, nil
}

// fileOne fetches the one scan the answer names and files it.
func (s *Service) fileOne(ctx context.Context, ans map[string]any) (map[string]any, error) {
	key, _ := ans["key"].(string)
	sum, _ := ans["sha256"].(string)
	link, _ := ans["link"].(string)
	if key == "" || sum == "" || link == "" {
		return nil, fmt.Errorf("the device process's answer to the scan names no stored scan or no link to it: %v", withoutLink(ans))
	}
	// The scan is read from the store by the link the device process signed, never through the bus (ADR 0468 §4), and
	// must hash to the scan's SHA-256 before anything is written.
	var buf bytes.Buffer
	gctx, cancel := context.WithTimeout(ctx, 5*time.Minute)
	defer cancel()
	if _, err := objectlink.Get(gctx, s.fetchClient(), link, &buf, objectlink.MaxDocument, sum); err != nil {
		return nil, fmt.Errorf("the scan is kept as %s, but it could not be read from the store: %w", key, err)
	}
	pdf := buf.Bytes()
	at, err := time.Parse(time.RFC3339, fmt.Sprint(ans["scanned-at"]))
	if err != nil {
		at = s.Now()
	}
	path, err := File(s.Settings.Folder, pdf, at.Local())
	if err != nil {
		return nil, fmt.Errorf("the scan is kept as %s, but could not be filed: %w", key, err)
	}
	body := map[string]any{}
	for k, v := range ans {
		if k != "key" && k != "size" && k != "link" && k != "link-seconds" {
			body[k] = v
		}
	}
	body["path"] = path
	return body, nil
}

// withoutLink is an answer said in an error without its link, which carries a signature.
func withoutLink(ans map[string]any) map[string]any {
	out := map[string]any{}
	for k, v := range ans {
		if k != "link" {
			out[k] = v
		}
	}
	return out
}

func (s *Service) fetchClient() *http.Client {
	if s.Fetch != nil {
		return s.Fetch
	}
	return &http.Client{}
}

func (s *Service) docExpiry() time.Duration {
	if s.DocExpiry > 0 {
		return s.DocExpiry
	}
	return 30 * time.Minute
}

func (s *Service) docRetry() time.Duration {
	if s.DocRetry > 0 {
		return s.DocRetry
	}
	return 5 * time.Minute
}

// fetchClientDefault is what a Service made by the tools reads the links with; a test puts a client of its own in its place before it makes the tools.
var fetchClient = &http.Client{}

// ScanStatus says whether the scanner answers, and what it is doing. A scanner that does not answer is an
// answer, not an error: that is how the module's health is read.
func (s *Service) ScanStatus(ctx context.Context) (map[string]any, error) {
	if err := s.Settings.Missing(); err != nil {
		return nil, err
	}
	st, err := NewClient(s.Settings.Scanner).Status(ctx)
	if err != nil {
		return map[string]any{"answers": false, "why": err.Error()}, nil
	}
	return map[string]any{"answers": true, "state": st.State, "adf-state": st.AdfState}, nil
}

// ScanList is the newest filed scans, newest first.
func (s *Service) ScanList(limit int) (map[string]any, error) {
	if err := s.Settings.Missing(); err != nil {
		return nil, err
	}
	scans, err := List(s.Settings.Folder, limit)
	if err != nil {
		return nil, err
	}
	return map[string]any{"folder": s.Settings.Folder, "scans": scans}, nil
}

// armDocument closes the open document in d, from the expiry on: if that fails (the device process dropped the ask,
// the bucket was away) it is tried again after docRetry, up to ten times, unless the document is gone or closed
// already. A document is never forgotten while it can still be filed.
func (s *Service) armDocument(doc string, d time.Duration, attempts int) {
	s.Jobs.arm(doc, d, func() {
		_, j, err := s.startScan(Ask{Op: OpScan, Document: doc, Finish: true}, scanRun)
		if err == nil {
			<-j.done
			err = j.err
		}
		if err == nil || attempts >= 10 || strings.Contains(err.Error(), "is closed already") || strings.Contains(err.Error(), "its id is wrong") {
			return
		}
		s.armDocument(doc, s.docRetry(), attempts+1)
	})
}

// Recover arms the expiry of every document the device process finds open in the bucket, so that a restart of
// this process forgets none (issue 1538). A document gets what is left of its time since its last page.
func (s *Service) Recover(ctx context.Context) error {
	ans, err := s.call(ctx, Ask{Op: OpOpen}, s.AnswerWait)
	if err != nil {
		return err
	}
	open, _ := ans["open"].([]any)
	for _, o := range open {
		m, _ := o.(map[string]any)
		id, _ := m["id"].(string)
		if id == "" {
			continue
		}
		left := s.docExpiry()
		if at, err := time.Parse(time.RFC3339, fmt.Sprint(m["updated"])); err == nil {
			left = max(s.docExpiry()-s.Now().Sub(at), time.Second)
		}
		s.armDocument(id, left, 0)
	}
	return nil
}

// tryRecover says whether this process has still to recover its open documents, and takes the turn; failed gives it
// back.
func (j *Jobs) tryRecover() bool {
	j.mu.Lock()
	defer j.mu.Unlock()
	if j.recovered {
		return false
	}
	j.recovered = true
	return true
}

func (j *Jobs) recoverFailed() {
	j.mu.Lock()
	j.recovered = false
	j.mu.Unlock()
}