sm-seo-plugin/classes/crawl/job.go

package crawl

import (
	"context"
	"errors"
	"strconv"
	"strings"
	"time"

	"git.golem15.com/golem15/sm-seo-plugin/classes"
	"git.golem15.com/golem15/sm-seo-plugin/classes/schema"
	"git.golem15.com/golem15/sm-seo-plugin/models"
	"git.golem15.com/golem15/sm-sitemap-plugin/contract"
	"git.golem15.com/golem15/summercms/modules/conga"
	"git.golem15.com/golem15/summercms/modules/diary"
	"git.golem15.com/golem15/summercms/modules/lagoon"
	"git.golem15.com/golem15/summercms/modules/pact"
	"gorm.io/gorm"
)

// SitemapTool is ToolSitemapSettings while the application publishes the
// sitemap plugin's contract.Listing (the sitemap plugin is mounted), else
// "".
func SitemapTool(svc *classes.Service) string {
	app := svc.App()
	if app == nil {
		return ""
	}
	if _, ok := app.Lookup[contract.Listing](); ok {
		return ToolSitemapSettings
	}
	return ""
}

// CrawlKind is the River kind of the crawl job.
const CrawlKind = "golem15.seo.crawl"

// CrawlArgs are the arguments of the crawl job: the crawl row it runs.
type CrawlArgs struct {
	CrawlID uint `json:"crawl_id"`
}

// Kind is CrawlKind.
func (CrawlArgs) Kind() string { return CrawlKind }

// The crawl job's time bounds (P33-60). JobTimeout is the job's deadline;
// FinishTimeout bounds storing what a crawl found, on a context the job's
// cancellation cannot reach; the final status is then written on a
// 30-second bound of its own. The job writes heartbeat_at every
// HeartbeatEvery, and a running crawl without a heartbeat for StaleAfter
// is ended by the next Start or delete (RecoverStale).
const (
	JobTimeout     = 6 * time.Hour
	FinishTimeout  = 10 * time.Minute
	HeartbeatEvery = time.Minute
	StaleAfter     = 10 * time.Minute
)

// heartbeatWrite bounds one heartbeat update.
const heartbeatWrite = 10 * time.Second

// finalWriteTimeout bounds a crawl's final status write and the end of its
// conga record, on a context of their own, so a store that used up its
// FinishTimeout bound still ends the crawl (33-REVIEW-GAPFIX WR-03).
const finalWriteTimeout = 30 * time.Second

type finishTimeoutKey struct{}

// WithFinishTimeout returns a context whose crawls bound storing their
// results by d instead of FinishTimeout. It is the test seam of the store
// bound, settable only from Go code (like WithRenderer and
// fetchguard.WithTransport): no config key, setting, request or MCP call
// reaches it.
func WithFinishTimeout(ctx context.Context, d time.Duration) context.Context {
	return context.WithValue(ctx, finishTimeoutKey{}, d)
}

// finishTimeoutFrom returns the store bound of ctx: its WithFinishTimeout
// value when that is positive, else FinishTimeout.
func finishTimeoutFrom(ctx context.Context) time.Duration {
	if ctx != nil {
		if d, ok := ctx.Value(finishTimeoutKey{}).(time.Duration); ok && d > 0 {
			return d
		}
	}
	return FinishTimeout
}

// The crawl limits used until the Site audit settings say otherwise.
const (
	DefaultMaxPages = 5000
	DefaultDelay    = 250 * time.Millisecond
)

// AlreadyRunningError refuses a start while a crawl is queued or running
// (P33-39): one crawl runs at a time.
type AlreadyRunningError struct {
	Crawl models.Crawl
}

func (e *AlreadyRunningError) Error() string { return "golem15.seo: a crawl is already running" }

// Since is the time the running crawl started, else its creation time.
func (e *AlreadyRunningError) Since() time.Time {
	if e.Crawl.StartedAt != nil {
		return *e.Crawl.StartedAt
	}
	if e.Crawl.CreatedAt != nil {
		return *e.Crawl.CreatedAt
	}
	return time.Time{}
}

// Start queues a crawl (D-14, P33-39) inside tx: the settings row is
// locked so two starts cannot both pass the check that no crawl is queued
// or running, a running crawl whose job stopped writing its heartbeat is
// ended (RecoverStale), then the crawl row is written and the conga job
// dispatched in the same transaction. A missing site address refuses with
// ErrNoSiteURL, a queued or running crawl (also one whose cancel is
// requested but whose job has not stopped) with *AlreadyRunningError.
func Start(ctx context.Context, svc *classes.Service, tx *gorm.DB, trigger, name string) (models.Crawl, error) {
	site := svc.SiteURL()
	if site == "" {
		return models.Crawl{}, ErrNoSiteURL
	}
	if _, err := classes.SettingsRow(tx); err != nil {
		return models.Crawl{}, err
	}
	if err := tx.Exec(`SELECT id FROM golem15_seo_settings WHERE id = 1 FOR UPDATE`).Error; err != nil {
		return models.Crawl{}, err
	}
	stale, err := RecoverStale(tx, time.Now().UTC())
	if err != nil {
		return models.Crawl{}, err
	}
	endStaleJobs(ctx, svc, tx, stale)
	active, err := ActiveCrawl(tx)
	if err != nil {
		return models.Crawl{}, err
	}
	if active != nil {
		return models.Crawl{}, &AlreadyRunningError{Crawl: *active}
	}
	limits, err := LoadLimits(tx)
	if err != nil {
		return models.Crawl{}, err
	}
	now := time.Now().UTC()
	row := models.Crawl{
		Status: models.CrawlQueued, Trigger: trigger, TriggerName: name, Site: site,
		MaxPages: limits.MaxPages, RenderStatus: models.RenderSkipped, CreatedAt: &now,
	}
	if err := tx.Create(&row).Error; err != nil {
		return models.Crawl{}, err
	}
	m, err := conga.From(svc.App())
	if err != nil {
		return models.Crawl{}, err
	}
	id, err := m.Dispatch(ctx, tx, CrawlArgs{CrawlID: row.ID}, conga.DispatchOpts{Label: svc.Phrase("golem15.seo::lang.jobs.crawl", nil)})
	if err != nil {
		return models.Crawl{}, err
	}
	row.JobID = &id
	if err := tx.Model(&models.Crawl{}).Where("id = ?", row.ID).Update("job_id", id).Error; err != nil {
		return models.Crawl{}, err
	}
	return row, nil
}

// Limits are a crawl's politeness limits and the Site audit settings it
// reads: pages per crawl, the pause between requests, the crawls kept and
// the generic link texts.
type Limits struct {
	MaxPages       int
	Delay          time.Duration
	Keep           int
	GenericAnchors []string
}

// DefaultKeep is the number of crawls kept.
const DefaultKeep = 10

// LoadLimits reads the crawl limits from the Site audit settings; a value
// outside its range (written around the settings screen) falls back to the
// default.
func LoadLimits(db *gorm.DB) (Limits, error) {
	st, err := classes.LoadSettings(db)
	if err != nil {
		return Limits{}, err
	}
	out := Limits{MaxPages: DefaultMaxPages, Delay: DefaultDelay, Keep: DefaultKeep, GenericAnchors: ParseAnchors(st.GenericAnchors)}
	if st.CrawlMaxPages >= models.MinCrawlMaxPages && st.CrawlMaxPages <= models.MaxCrawlMaxPages {
		out.MaxPages = st.CrawlMaxPages
	}
	if st.CrawlDelay >= models.MinCrawlDelay && st.CrawlDelay <= models.MaxCrawlDelay {
		out.Delay = time.Duration(st.CrawlDelay) * time.Millisecond
	}
	if st.CrawlKeep >= models.MinCrawlKeep && st.CrawlKeep <= models.MaxCrawlKeep {
		out.Keep = st.CrawlKeep
	}
	return out, nil
}

// LoadCrawl reads one crawl.
func LoadCrawl(db *gorm.DB, id uint) (models.Crawl, error) {
	var rows []models.Crawl
	if err := db.Where("id = ?", id).Limit(1).Find(&rows).Error; err != nil {
		return models.Crawl{}, err
	}
	if len(rows) == 0 {
		return models.Crawl{}, gorm.ErrRecordNotFound
	}
	return rows[0], nil
}

// Run runs one queued crawl: it fetches the site, checks every page,
// stores the pages, links and findings with the CMS-data issues, and marks
// the crawl finished, cancelled (pages and issues fetched so far are kept)
// or failed (with the error; what was fetched before the failure is kept).
// A crawl that is no longer queued is left alone.
//
// The job's context ends on a cancel (River cancels a running job's
// context) or at the job timeout; what was fetched is then stored on a
// context detached from it with its own FinishTimeout bound, and the final
// status is written on another detached context bounded by
// finalWriteTimeout, so a cancelled or timed-out crawl keeps its results
// and a crawl whose store ran out of time still ends: none stays running.
// While it runs, the job writes heartbeat_at every HeartbeatEvery.
func Run(ctx context.Context, svc *classes.Service, id uint) error {
	gdb, err := svc.GormDB()
	if err != nil {
		return err
	}
	db := gdb.WithContext(ctx)
	row, err := LoadCrawl(db, id)
	if err != nil {
		return err
	}
	if row.Status != models.CrawlQueued {
		return nil
	}
	now := time.Now().UTC()
	moved := db.Model(&models.Crawl{}).Where("id = ? AND status = ?", id, models.CrawlQueued).
		Updates(map[string]any{"status": models.CrawlRunning, "started_at": now, "heartbeat_at": now})
	if moved.Error != nil {
		return moved.Error
	}
	if moved.RowsAffected == 0 {
		return nil
	}
	stop := heartbeat(ctx, gdb, id)
	defer stop()
	res, runErr := run(ctx, svc, db, row)
	return finish(context.WithoutCancel(ctx), svc, gdb, row, res, runErr, ctx.Err(), finishTimeoutFrom(ctx))
}

// heartbeat writes the crawl's heartbeat_at every HeartbeatEvery until the
// returned stop is called; each write runs on a context detached from the
// job's, bounded by heartbeatWrite.
func heartbeat(ctx context.Context, gdb *gorm.DB, id uint) (stop func()) {
	done := make(chan struct{})
	finished := make(chan struct{})
	go func() {
		defer close(finished)
		t := time.NewTicker(HeartbeatEvery)
		defer t.Stop()
		for {
			select {
			case <-done:
				return
			case <-t.C:
				hctx, cancel := context.WithTimeout(context.WithoutCancel(ctx), heartbeatWrite)
				_ = gdb.WithContext(hctx).Model(&models.Crawl{}).Where("id = ? AND status = ?", id, models.CrawlRunning).
					Update("heartbeat_at", time.Now().UTC()).Error
				cancel()
			}
		}
	}()
	return func() {
		close(done)
		<-finished
	}
}

// run fetches and checks the site of one crawl.
func run(ctx context.Context, svc *classes.Service, db *gorm.DB, row models.Crawl) (*Result, error) {
	f, err := NewFetcher(row.Site, svc.Config())
	if err != nil {
		return nil, err
	}
	limits, err := LoadLimits(db)
	if err != nil {
		return nil, err
	}
	if row.MaxPages > 0 {
		limits.MaxPages = row.MaxPages
	}
	var m *conga.Manager
	if app := svc.App(); app != nil {
		m, _ = conga.From(app)
	}
	// The rendered comparison: the test seam's renderer, else Chrome from
	// golem15.seo.crawl_chrome_path (none: skipped). Chrome is closed when
	// the crawl finishes, fails or is cancelled.
	renderer, renderErr := rendererFrom(ctx), error(nil)
	if renderer == nil {
		renderer, renderErr = NewRenderer(ctx, svc.Config(), f.Site(), f.IsTrusted(), f.UA())
		if renderer != nil {
			defer func() { _ = renderer.Close() }()
		}
	}
	opt := Options{
		MaxPages:  limits.MaxPages,
		Delay:     limits.Delay,
		Renderer:  renderer,
		RenderErr: renderErr,
		RenderMax: RenderMax(svc.Config()),
		Canceled: func(ctx context.Context) bool {
			if cancelRequested(db, row.ID) {
				return true
			}
			if m != nil && row.JobID != nil {
				if c, err := m.CheckIfCanceled(ctx, *row.JobID); err == nil && c {
					return true
				}
			}
			return false
		},
		Progress: func(ctx context.Context, n int) {
			_ = db.Model(&models.Crawl{}).Where("id = ?", row.ID).Update("pages", n).Error
			if m != nil && row.JobID != nil {
				_ = m.UpdateJobState(ctx, *row.JobID, n, nil)
			}
		},
	}
	if m != nil && row.JobID != nil {
		_ = m.StartJob(ctx, *row.JobID, limits.MaxPages)
	}
	return Crawl(ctx, f, opt)
}

// cancelRequested reports whether a cancel of the crawl was requested: its
// cancel_requested_at is set, or its status is cancelled.
func cancelRequested(db *gorm.DB, id uint) bool {
	var rows []struct {
		Status    string
		Requested bool
	}
	if err := db.Raw(`SELECT status, cancel_requested_at IS NOT NULL AS requested FROM golem15_seo_crawls WHERE id = ?`, id).Scan(&rows).Error; err != nil || len(rows) == 0 {
		return false
	}
	return rows[0].Requested || rows[0].Status == models.CrawlCancelled
}

// timeLimitError is the error of a crawl that reached JobTimeout.
func timeLimitError() string {
	limit := JobTimeout.String()
	if JobTimeout%time.Hour == 0 {
		limit = strconv.Itoa(int(JobTimeout/time.Hour)) + "h"
	}
	return "the crawl reached its time limit of " + limit
}

// finalStatus decides a crawl's final status and error: a requested
// cancel, or a cancelled job context, ends cancelled; the job's deadline
// fails it with the time-limit error; another run error fails it with that
// error; else it finished (cancelled when the crawl stopped on a cancel).
func finalStatus(requested bool, res *Result, runErr, ctxErr error) (string, string) {
	switch {
	case requested:
		return models.CrawlCancelled, ""
	case errors.Is(ctxErr, context.DeadlineExceeded) || errors.Is(runErr, context.DeadlineExceeded):
		return models.CrawlFailed, timeLimitError()
	case errors.Is(ctxErr, context.Canceled) || errors.Is(runErr, context.Canceled):
		return models.CrawlCancelled, ""
	case runErr != nil:
		return models.CrawlFailed, runErr.Error()
	case res != nil && res.Canceled:
		return models.CrawlCancelled, ""
	}
	return models.CrawlFinished, ""
}

// finish stores what a crawl found and writes its final state. parent is
// Run's context detached from the job's cancellation, gdb the undecorated
// database, ctxErr the job context's error (nil while it was alive). The
// cancel check, the limits and the store run on a context bounded by
// storeBound; a store that runs past it fails the crawl with an error that
// names the bound and the store error. The final status write
// and the end of the conga record then run on a fresh context bounded by
// finalWriteTimeout, and Prune on one of its own. A LoadLimits failure
// falls back to the default limits.
func finish(parent context.Context, svc *classes.Service, gdb *gorm.DB, row models.Crawl, res *Result, runErr, ctxErr error, storeBound time.Duration) error {
	storeCtx, cancelStore := context.WithTimeout(parent, storeBound)
	defer cancelStore()
	db := gdb.WithContext(storeCtx)
	status, errText := finalStatus(cancelRequested(db, row.ID), res, runErr, ctxErr)
	limits, err := LoadLimits(db)
	if err != nil {
		pluginLog.Warn("golem15.seo: crawl limits, using the defaults", "crawl", row.ID, "error", err)
		limits = Limits{MaxPages: DefaultMaxPages, Delay: DefaultDelay, Keep: DefaultKeep}
	}
	store := func(ctx context.Context, tx *gorm.DB) error {
		if res == nil {
			return nil
		}
		if err := SavePages(tx, row.ID, res); err != nil {
			return err
		}
		keyphrases, err := Keyphrases(tx)
		if err != nil {
			return err
		}
		findings := Check(res, &Env{Lookup: Lookup(ctx, svc, tx), Staging: svc.Staging(), GenericAnchors: limits.GenericAnchors, Keyphrases: keyphrases, SitemapTool: SitemapTool(svc)})
		cms, err := svc.Issues(classes.WithTx(ctx, tx))
		if err != nil {
			return err
		}
		id := row.ID
		return SaveIssues(tx, &id, findings, cms)
	}
	if err := lagoon.Transaction(storeCtx, db, store); err != nil {
		status, errText = models.CrawlFailed, err.Error()
		if errors.Is(storeCtx.Err(), context.DeadlineExceeded) {
			errText = "storing the crawl results took longer than " + storeBound.String() + ": " + err.Error()
		}
	}
	ctx, cancelWrite := context.WithTimeout(parent, finalWriteTimeout)
	defer cancelWrite()
	updates := map[string]any{"status": status, "error": errText, "finished_at": time.Now().UTC()}
	if res != nil {
		updates["pages"] = len(res.Order)
		updates["limit_hit"] = res.LimitHit
		updates["render_status"] = res.Render.Status
		updates["render_pages"] = res.Render.Pages
		updates["render_note"] = res.Render.Note
	}
	if err := gdb.WithContext(ctx).Model(&models.Crawl{}).Where("id = ?", row.ID).Updates(updates).Error; err != nil {
		return err
	}
	if m, err := conga.From(svc.App()); err == nil && row.JobID != nil {
		meta := map[string]any{"crawl_id": row.ID, "status": status}
		switch status {
		case models.CrawlFailed:
			_ = m.FailJob(ctx, *row.JobID, meta)
		case models.CrawlCancelled:
			_ = m.StopJob(ctx, *row.JobID, meta)
		default:
			_ = m.CompleteJob(ctx, *row.JobID, meta)
		}
	}
	if status == models.CrawlFailed {
		pluginLog.Warn("golem15.seo: crawl failed", "crawl", row.ID, "error", errText)
	}
	// Prune deletes old crawls with their pages and links, so it keeps a
	// store-sized bound; it runs after the final state is written, so it
	// never holds a crawl running.
	pruneCtx, cancelPrune := context.WithTimeout(parent, storeBound)
	defer cancelPrune()
	if err := Prune(gdb.WithContext(pruneCtx), limits.Keep); err != nil {
		pluginLog.Warn("golem15.seo: prune crawls", "error", err)
	}
	return nil
}

// endStaleJobs fails or stops the conga jobs of crawls RecoverStale ended,
// once tx commits; best effort.
func endStaleJobs(ctx context.Context, svc *classes.Service, tx *gorm.DB, rows []models.Crawl) {
	if len(rows) == 0 {
		return
	}
	lagoon.AfterCommit(ctx, tx, func(ctx context.Context, _ *gorm.DB) {
		m, err := conga.From(svc.App())
		if err != nil {
			return
		}
		for _, r := range rows {
			if r.JobID == nil {
				continue
			}
			meta := map[string]any{"crawl_id": r.ID, "status": r.Status}
			if r.Status == models.CrawlCancelled {
				_ = m.StopJob(ctx, *r.JobID, meta)
			} else {
				_ = m.FailJob(ctx, *r.JobID, meta)
			}
		}
	})
}

// Lookup returns the CMS record behind a crawled address (cached per
// address): the content type whose Match answers the path exactly, its
// descriptor facts in the matched locale, its primary parent's address and
// whether its SEO tab stores custom JSON-LD.
func Lookup(ctx context.Context, svc *classes.Service, db *gorm.DB) func(key string) *Record {
	cache := map[string]*Record{}
	reg := svc.Registry()
	unprefixed := svc.UnprefixedLocale()
	return func(key string) *Record {
		if reg == nil {
			return nil
		}
		path, _, _ := strings.Cut(key, "?")
		if r, ok := cache[path]; ok {
			return r
		}
		cache[path] = nil
		typeKey, match, err := reg.MatchPath(ctx, path, unprefixed)
		if err != nil || match == nil || !match.Exact {
			return nil
		}
		typ, ok := reg.ContentType(typeKey)
		if !ok || typ.Content == nil {
			return nil
		}
		rec, err := reg.Describe(ctx, typeKey, match.ID, unprefixed)
		if err != nil || rec == nil {
			return nil
		}
		out := &Record{Type: typeKey, ID: match.ID, Locale: match.Locale, TopHeading: rec.TopHeading, Tools: typ.Content.Tools}
		var loc *pact.ContentLocale
		for i := range rec.Locales {
			if rec.Locales[i].Locale == match.Locale {
				loc = &rec.Locales[i]
				break
			}
		}
		if loc == nil && len(rec.Locales) > 0 {
			loc = &rec.Locales[0]
		}
		if loc != nil {
			out.Locale = loc.Locale
			out.Title, out.Words, out.Headings, out.Images = loc.Title, loc.Words, loc.Headings, loc.Images
		}
		if len(rec.Parents) > 0 {
			if parent, err := reg.Describe(ctx, rec.Parents[0].Type, rec.Parents[0].ID, unprefixed); err == nil && parent != nil {
				for _, pl := range parent.Locales {
					if pl.Locale == out.Locale && pl.URL != "" {
						out.Links = append(out.Links, pl.URL)
					}
				}
			}
		}
		if entry, found, err := classes.LoadEntry(db, typ.Content.Model, match.ID, out.Locale); err == nil && found && entry.Data != nil {
			if raw, _ := entry.Data["schema"].(string); raw != "" {
				if v, err := schema.ParseValue(raw); err == nil && strings.TrimSpace(v.Custom) != "" {
					out.CustomJSONLD = true
				}
			}
		}
		cache[path] = out
		return out
	}
}

// Keyphrases maps each address with an SEO entry to its focus keyphrase.
func Keyphrases(db *gorm.DB) (map[string]string, error) {
	var rows []struct {
		URL       string
		Keyphrase string
	}
	if err := db.Raw(`SELECT url, keyphrase FROM golem15_seo_entries WHERE url <> '' AND keyphrase <> ''`).Scan(&rows).Error; err != nil {
		return nil, err
	}
	out := make(map[string]string, len(rows))
	for _, r := range rows {
		out[r.URL] = r.Keyphrase
	}
	return out, nil
}

// errCancelledCrawl refuses cancelling a crawl that is not queued or
// running.
var errCancelledCrawl = errors.New("golem15.seo: the crawl is not running")

// Cancel stops a queued or running crawl inside tx. A queued crawl ends
// cancelled at once (by name, finished_at set). A running crawl gets
// cancel_requested_at and cancelled_by and stays running: its job stops
// between requests and writes cancelled with the pages and issues fetched
// so far, and until then no other crawl starts. The conga job is cancelled
// in both cases.
func Cancel(ctx context.Context, svc *classes.Service, tx *gorm.DB, row models.Crawl, name string) error {
	if !row.Active() {
		return errCancelledCrawl
	}
	now := time.Now().UTC()
	queued := tx.Model(&models.Crawl{}).Where("id = ? AND status = ?", row.ID, models.CrawlQueued).
		Updates(map[string]any{"status": models.CrawlCancelled, "cancelled_by": name, "finished_at": now})
	if queued.Error != nil {
		return queued.Error
	}
	if queued.RowsAffected == 0 {
		running := tx.Model(&models.Crawl{}).Where("id = ? AND status = ?", row.ID, models.CrawlRunning).
			Updates(map[string]any{"cancel_requested_at": now, "cancelled_by": name})
		if running.Error != nil {
			return running.Error
		}
		if running.RowsAffected == 0 {
			return errCancelledCrawl
		}
	}
	if row.JobID != nil {
		if m, err := conga.From(svc.App()); err == nil {
			if err := m.CancelJob(ctx, *row.JobID); err != nil {
				pluginLog.Warn("golem15.seo: cancel crawl job", "crawl", row.ID, "error", err)
			}
		}
	}
	return nil
}

// ErrNotRunning is a cancel of a crawl that is not queued or running.
var ErrNotRunning = errCancelledCrawl

// pluginLog is the crawler's logger (component seo).
var pluginLog = diary.For("seo")

// ScheduledDue reports whether the nightly entry starts a crawl on day
// (in the app's location) for the crawl_schedule setting: daily every
// night, weekly on Mondays, never when off (P33-40).
func ScheduledDue(schedule string, day time.Time) bool {
	switch schedule {
	case models.CrawlDaily:
		return true
	case models.CrawlWeekly:
		return day.Weekday() == time.Monday
	}
	return false
}