Work that outlives the request

go func() { processa(docID) }()

trilha add tasks --dry-run

  + internal/trabalho/trabalho.go
  + internal/trabalho/trabalho_test.go
  + app/tarefas/page.go
  + tarefas_test.go
  ~ app/setup.go (one line added)

--dry-run: nothing was written

// Novo builds the engine with the handler already registered. The pace of the
// stages is read once, here, so each engine carries its own — a test asks for
// a fast one through the environment, before the app is built, and nothing is
// shared afterwards.
func Novo(hooks *webhook.Hooks) *task.Tasks {
	m := &motor{passo: 250 * time.Millisecond, hooks: hooks}
	if v := os.Getenv("BLOG_TASK_STEP"); v != "" {
		if d, err := time.ParseDuration(v); err == nil && d > 0 {
			m.passo = d
		}
	}
	t := task.New(task.Options{Workers: 2})
	t.Handle(Nome, m.processar)
	return t
}

	motor := tarefas.Novo(hooks)
	trilha.Provide(a, motor)
	if err := motor.Setup(a); err != nil {
		return err
	}

// POST starts the processing of one document and goes to its screen.
func POST(c *trilha.Ctx) error {
	id := c.Form("documento")
	if _, ok := documentos.Um(id); !ok {
		return trilha.Errorf(http.StatusNotFound, "documento não encontrado")
	}
	motor := trilha.Use[*task.Tasks](c)
	// Duplo clique no botão devolve o id da que já está indo, e não uma
	// segunda: é o dedupe por chave, e é por isso que a chave é o documento.
	tarefa, err := motor.Run(c, tarefas.Nome, id)
	if err != nil {
		return err
	}
	return c.Redirect("/tarefas/" + tarefa)
}

// processar is the work. It reports each stage with Progress.Step and looks at
// the context between them: a task that never looks is a task the shutdown has
// to cancel by force.
func (m *motor) processar(ctx context.Context, p *task.Progress) error {
	doc, ok := documentos.Um(p.Key)
	if !ok {
		return errors.New("documento não existe mais")
	}
	documentos.Marcar(p.Key, "processando")

	estagios := []string{"Extrair texto", "Classificar", "Indexar", "Concluir"}
	for i, estagio := range estagios {
		p.Step(estagio, i+1, len(estagios))
		select {
		case <-ctx.Done():
			// Interrupted is not failed, and the difference is what the screen
			// tells somebody before they press "try again".
			documentos.Marcar(p.Key, "fila")
			return ctx.Err()
		case <-time.After(m.passo):
		}
		// Um documento com "erro" no nome falha de propósito: é o caminho que
		// o exemplo precisa mostrar tanto quanto o que dá certo.
		if i == 1 && strings.Contains(strings.ToLower(doc.Nome), "erro") {
			documentos.Marcar(p.Key, "fila")
			m.avisa("documento.falhou", doc)
			return errors.New("não consegui classificar: o arquivo não tem texto")
		}
	}
	documentos.Marcar(p.Key, "pronto")
	m.avisa("documento.processado", doc)
	return nil
}

// Page draws the progress, and only the progress when asked for the fragment.
func Page(c *trilha.Ctx) (h.Node, error) {
	motor := trilha.Use[*task.Tasks](c)
	id := c.Param("id")
	if c.Fragment() == "trilha-task" {
		return ui.TaskProgress(c, motor, id), nil
	}
	t, err := motor.Get(c.Context(), id)
	if err != nil {
		return nil, trilha.Errorf(http.StatusNotFound, "tarefa não encontrada")
	}
	c.SetTitle("Processando " + t.Key)
	return h.Div(
		ui.PageHeader("Processando "+t.Key, ui.Back{Href: "/tarefas", Label: "Tarefas"}),
		ui.Muted(h.Text("Esta página se atualiza sozinha e para quando a tarefa termina.")),
		ui.Card(ui.CardContent(ui.TaskProgress(c, motor, id))),
		h.P(h.A(h.Href("/tarefas"), h.Text("Voltar para as tarefas"))),
		ui.LiveScript(c),
	), nil
}

// POST runs the task again and goes to the new run's screen.
func POST(c *trilha.Ctx) error {
	motor := trilha.Use[*task.Tasks](c)
	novo, err := motor.Retry(c, c.Form("id"))
	if err != nil {
		return trilha.Errorf(http.StatusNotFound, "tarefa não encontrada")
	}
	c.Flash("info", "Rodando de novo.")
	return c.Redirect("/tarefas/" + novo)
}

// TaskSchema is the table. Two indexes and both are load-bearing: the first is
// the deduplication, which runs on every Run, and the second is the listing,
// which is the only order the screen ever asks for.
const TaskSchema = `
CREATE TABLE IF NOT EXISTS trilha_tasks (
	id       TEXT PRIMARY KEY,
	name     TEXT NOT NULL,
	key      TEXT NOT NULL,
	state    TEXT NOT NULL,
	step     TEXT NOT NULL DEFAULT '',
	n        INTEGER NOT NULL DEFAULT 0,
	of_n     INTEGER NOT NULL DEFAULT 0,
	err      TEXT NOT NULL DEFAULT '',
	by_whom  TEXT NOT NULL DEFAULT '',
	queued   TIMESTAMPTZ NOT NULL,
	started  TIMESTAMPTZ,
	ended    TIMESTAMPTZ
);
CREATE INDEX IF NOT EXISTS trilha_tasks_active ON trilha_tasks (name, key) WHERE state IN ('queued','running');
CREATE INDEX IF NOT EXISTS trilha_tasks_recent ON trilha_tasks (queued DESC);
`

// Active is the deduplication, and it is the one query that runs on every
// Run: the partial index above exists for this line.
func (s TaskSQL) Active(ctx context.Context, name, key string) (string, error) {
	var id string
	err := s.DB.QueryRowContext(ctx,
		`SELECT id FROM trilha_tasks WHERE name = $1 AND key = $2 AND state IN ('queued','running') LIMIT 1`,
		name, key).Scan(&id)
	if errors.Is(err, sql.ErrNoRows) {
		return "", nil
	}
	return id, err
}

// Interrupt runs once, at boot, and is the reason a table beats memory here:
// it is the row that says the deploy caught this one halfway, instead of a
// screen waiting forever for a process that is gone.
func (s TaskSQL) Interrupt(ctx context.Context, at time.Time) (int, error) {
	res, err := s.DB.ExecContext(ctx, `
		UPDATE trilha_tasks SET state = 'interrupted', ended = $1,
			err = 'the process stopped before this finished'
		WHERE state IN ('queued','running')`, at)
	if err != nil {
		return 0, err
	}
	n, err := res.RowsAffected()
	return int(n), err
}

	id, _ := primeiroDoc(t)
	inicio := time.Now()
	rec := c.PostForm("/tarefas", url.Values{"documento": {id}})
	rec.WantStatus(http.StatusSeeOther)
	if passou := time.Since(inicio); passou > time.Second {
		t.Fatalf("o POST esperou %s: a tarefa rodou dentro da requisição", passou)
	}