PrintToVeraeTimeStamp/internal/netprint/events.go

package netprint

import (
	"encoding/json"
	"fmt"
	"log"
	"net/http"
	"strconv"
	"strings"
	"sync"
	"time"
)

// JobEvent is one completed job, told to the menu-bar app over the loopback
// control API. The daemon does not open a share sheet.
type JobEvent struct {
	Seq            int64     `json:"seq"`
	Kind           string    `json:"kind"`
	ID             int       `json:"id"`
	ReceiptID      string    `json:"receiptId"`
	Title          string    `json:"title"`
	AirDrop        bool      `json:"airdrop"`
	CertificateURL string    `json:"certificateUrl"`
	ReceiptJSONURL string    `json:"receiptJsonUrl"`
	At             time.Time `json:"at"`
}

type jobBus struct {
	mu     sync.Mutex
	next   int64
	recent []JobEvent
	subs   map[chan JobEvent]struct{}
	acked  map[int]bool
}

func (s *Server) afterJob(rec Record) {
	s.queueReceiptMail(rec)
	s.publishJobEvent(rec)
}

func (s *Server) publishJobEvent(rec Record) {
	if rec.State != "completed" {
		return
	}
	base := strings.TrimRight(s.controlURL, "/")
	ev := JobEvent{
		Kind:           "completed",
		ID:             rec.ID,
		ReceiptID:      rec.ReceiptID,
		Title:          rec.Title,
		AirDrop:        s.snapshot().AirDropOn(),
		CertificateURL: fmt.Sprintf("%s/jobs/%d/receipt.pdf", base, rec.ID),
		ReceiptJSONURL: fmt.Sprintf("%s/jobs/%d/receipt.json", base, rec.ID),
		At:             time.Now(),
	}
	s.jobs.publish(ev)
	if ev.AirDrop {
		log.Printf("netprint: job %d airdrop queued receipt=%s", rec.ID, rec.ReceiptID)
	}
}

func (b *jobBus) publish(ev JobEvent) {
	b.mu.Lock()
	if b.subs == nil {
		b.subs = map[chan JobEvent]struct{}{}
	}
	if b.acked == nil {
		b.acked = map[int]bool{}
	}
	b.next++
	ev.Seq = b.next
	b.recent = append(b.recent, ev)
	if len(b.recent) > 16 {
		b.recent = b.recent[len(b.recent)-16:]
	}
	subs := make([]chan JobEvent, 0, len(b.subs))
	for ch := range b.subs {
		subs = append(subs, ch)
	}
	b.mu.Unlock()
	for _, ch := range subs {
		select {
		case ch <- ev:
		default:
		}
	}
}

func (b *jobBus) subscribe(after int64) (backlog []JobEvent, ch chan JobEvent) {
	b.mu.Lock()
	defer b.mu.Unlock()
	if b.subs == nil {
		b.subs = map[chan JobEvent]struct{}{}
	}
	ch = make(chan JobEvent, 32)
	b.subs[ch] = struct{}{}
	if after == 0 {
		if ev, ok := b.pendingLocked(); ok {
			backlog = []JobEvent{ev}
		}
		return backlog, ch
	}
	for _, ev := range b.recent {
		if ev.Seq > after {
			backlog = append(backlog, ev)
		}
	}
	return backlog, ch
}

func (b *jobBus) unsubscribe(ch chan JobEvent) {
	b.mu.Lock()
	delete(b.subs, ch)
	b.mu.Unlock()
}

func (b *jobBus) ack(id int) {
	b.mu.Lock()
	defer b.mu.Unlock()
	if b.acked == nil {
		b.acked = map[int]bool{}
	}
	b.acked[id] = true
}

// pendingLocked is the newest AirDrop job the menu bar has not acknowledged.
// The caller holds b.mu.
func (b *jobBus) pendingLocked() (JobEvent, bool) {
	for i := len(b.recent) - 1; i >= 0; i-- {
		ev := b.recent[i]
		if !ev.AirDrop || b.acked[ev.ID] {
			continue
		}
		if time.Since(ev.At) > 10*time.Minute {
			continue
		}
		return ev, true
	}
	return JobEvent{}, false
}

func (s *Server) handleEvents(w http.ResponseWriter, r *http.Request) {
	if r.Method != http.MethodGet {
		http.Error(w, "method not allowed", http.StatusMethodNotAllowed)
		return
	}
	fl, ok := w.(http.Flusher)
	if !ok {
		http.Error(w, "stream unavailable", http.StatusInternalServerError)
		return
	}
	after, _ := strconv.ParseInt(r.URL.Query().Get("after"), 10, 64)
	w.Header().Set("Content-Type", "text/event-stream")
	w.Header().Set("Cache-Control", "no-store")
	w.Header().Set("Connection", "keep-alive")
	backlog, ch := s.jobs.subscribe(after)
	defer s.jobs.unsubscribe(ch)
	for _, ev := range backlog {
		writeJobEvent(w, ev)
	}
	fmt.Fprintf(w, ": ok\n\n")
	fl.Flush()
	tick := time.NewTicker(15 * time.Second)
	defer tick.Stop()
	for {
		select {
		case <-r.Context().Done():
			return
		case ev := <-ch:
			writeJobEvent(w, ev)
			fl.Flush()
		case <-tick.C:
			fmt.Fprintf(w, ": ping\n\n")
			fl.Flush()
		}
	}
}

func writeJobEvent(w http.ResponseWriter, ev JobEvent) {
	b, err := json.Marshal(ev)
	if err != nil {
		return
	}
	fmt.Fprintf(w, "event: job\ndata: %s\n\n", b)
}

func (s *Server) handleAirDropAck(w http.ResponseWriter, r *http.Request) {
	if r.Method != http.MethodPost {
		http.Error(w, "method not allowed", http.StatusMethodNotAllowed)
		return
	}
	var body struct {
		ID int `json:"id"`
	}
	if err := json.NewDecoder(http.MaxBytesReader(w, r.Body, 1<<20)).Decode(&body); err != nil || body.ID <= 0 {
		http.Error(w, "bad id", http.StatusBadRequest)
		return
	}
	s.jobs.ack(body.ID)
	writeJSON(w, map[string]bool{"ok": true})
}