package cooler import ( "bytes" "context" "encoding/json" "errors" "fmt" "io" "log/slog" "net" "net/http" "net/url" "strings" "sync" "time" "git.golem15.com/golem15/summercms/modules/backpack" "git.golem15.com/golem15/summercms/modules/lagoon" "gorm.io/gorm" ) const ( // SiteCachePurgeURLKey is the config key of the host site's cache purge // URL. Environment: SUMMER_SITE_CACHE__PURGE_URL. SiteCachePurgeURLKey = "site_cache.purge_url" // SiteCachePurgeTokenKey is the config key of the bearer token the purge // request carries. Environment: SUMMER_SITE_CACHE__PURGE_TOKEN. SiteCachePurgeTokenKey = "site_cache.purge_token" // SiteCachePurgeTimeout bounds one purge request. SiteCachePurgeTimeout = 2 * time.Second // SiteCachePurgeMinTokenLength is the shortest purge token accepted, the // same minimum the host site's purge route applies. SiteCachePurgeMinTokenLength = 32 // siteCacheAnswerLimit is how much of the purge answer is read. siteCacheAnswerLimit = 4 << 10 ) // SiteCachePurger asks the host site to drop its page caches after an admin // save. A nil *SiteCachePurger is disabled and does nothing. // // Purges queued by PurgeSiteCacheAfterCommit go through one worker per // purger: at most one request is in flight, and the saves committed while // it is sent collapse into one follow-up request carrying the last kind and // slug (each purge clears the site's whole route cache). type SiteCachePurger struct { url string endpoint string token string client *http.Client mu sync.Mutex // sending is true while the worker goroutine runs. sending bool // pending is the follow-up body, the latest one queued while sending. pending *siteCachePurgeBody // coalesced counts the queued bodies pending replaced. coalesced int // idle is closed when the current busy period ends; nil while idle. idle chan struct{} } // NewSiteCachePurger returns the purger for purgeURL and token, read from // site_cache.purge_url and site_cache.purge_token. Validation errors name // those keys; see NewSiteCachePurgerFor for the rules. func NewSiteCachePurger(purgeURL, token string) (*SiteCachePurger, error) { return NewSiteCachePurgerFor(SiteCachePurgeURLKey, SiteCachePurgeTokenKey, purgeURL, token) } // NewSiteCachePurgerFor returns the purger for purgeURL and token, read // from the config keys urlKey and tokenKey, which its errors name. With // both values empty the purge is disabled and it returns nil, nil. // Otherwise both must be set, the URL must be http or https with a host and // without user info or a fragment, an http URL must point at a loopback // host (127.0.0.0/8, ::1 or localhost) so the bearer token never crosses // the network in clear, and the token must hold at least // SiteCachePurgeMinTokenLength characters. An error never contains the // token. func NewSiteCachePurgerFor(urlKey, tokenKey, purgeURL, token string) (*SiteCachePurger, error) { purgeURL = strings.TrimSpace(purgeURL) if purgeURL == "" && token == "" { return nil, nil } if purgeURL == "" { return nil, fmt.Errorf("%s is set but %s is not; set both or neither", tokenKey, urlKey) } if token == "" { return nil, fmt.Errorf("%s is set but %s is not; set both or neither", urlKey, tokenKey) } u, err := url.Parse(purgeURL) if err != nil || (u.Scheme != "http" && u.Scheme != "https") || u.Host == "" || u.Hostname() == "" { return nil, fmt.Errorf("%s must be an http or https URL with a host", urlKey) } if u.User != nil { return nil, fmt.Errorf("%s must not carry user info", urlKey) } if u.Fragment != "" || strings.Contains(purgeURL, "#") { return nil, fmt.Errorf("%s must not carry a fragment", urlKey) } if u.Scheme == "http" && !loopbackHost(u.Hostname()) { return nil, fmt.Errorf("%s: an http URL must point at a loopback host (127.0.0.0/8, ::1 or localhost); use https for any other host", urlKey) } if len(token) < SiteCachePurgeMinTokenLength { return nil, fmt.Errorf("%s must hold at least %d characters", tokenKey, SiteCachePurgeMinTokenLength) } return &SiteCachePurger{url: u.String(), endpoint: u.Host + u.Path, token: token, client: &http.Client{CheckRedirect: noRedirect}}, nil } // noRedirect keeps the purge client on the configured URL: a 3xx answer is // returned as is (and logged as a failed purge with its status), so the // bearer token never follows a Location and the POST never becomes a GET. func noRedirect(*http.Request, []*http.Request) error { return http.ErrUseLastResponse } // Endpoint returns the purge URL's host and path, without the scheme, user // info or query, for log lines. func (p *SiteCachePurger) Endpoint() string { if p == nil { return "" } return p.endpoint } func loopbackHost(host string) bool { if strings.EqualFold(host, "localhost") { return true } ip := net.ParseIP(host) return ip != nil && ip.IsLoopback() } type siteCachePurgeBody struct { Kind string `json:"kind"` Slug string `json:"slug"` } // Purge sends the purge request for one saved entity: a POST of // {"kind":kind,"slug":slug} with the bearer token, bounded by // SiteCachePurgeTimeout and detached from ctx's cancellation. It blocks // until the request is done; it is the send the PurgeSiteCacheAfterCommit // queue uses. It logs the outcome and returns nothing: a purge never fails // a save. A redirect is never followed; a 3xx answer is logged as a failed // purge with its status. The token is never logged. func (p *SiteCachePurger) Purge(ctx context.Context, kind, slug string) { if p == nil { return } p.send(ctx, kind, slug, 0) } // send is Purge; coalesced, when above zero, is logged with the outcome: // the number of queued purges this request stands in for besides its own. func (p *SiteCachePurger) send(ctx context.Context, kind, slug string, coalesced int) { if ctx == nil { ctx = context.Background() } ctx, cancel := context.WithTimeout(context.WithoutCancel(ctx), SiteCachePurgeTimeout) defer cancel() log := slog.Default() failed := func(attrs ...any) { attrs = append([]any{"kind", kind, "slug", slug}, attrs...) if coalesced > 0 { attrs = append(attrs, "coalesced", coalesced) } log.Warn("cooler: site cache purge failed", attrs...) } raw, err := json.Marshal(siteCachePurgeBody{Kind: kind, Slug: slug}) if err != nil { failed("error", err.Error()) return } req, err := http.NewRequestWithContext(ctx, http.MethodPost, p.url, bytes.NewReader(raw)) if err != nil { failed("error", "invalid request") return } req.Header.Set("Authorization", "Bearer "+p.token) req.Header.Set("Content-Type", "application/json") req.Header.Set("Accept", "application/json") client := p.client if client == nil { client = &http.Client{CheckRedirect: noRedirect} } res, err := client.Do(req) if err != nil { failed("error", purgeTransportError(err)) return } defer res.Body.Close() body, _ := io.ReadAll(io.LimitReader(res.Body, siteCacheAnswerLimit)) if res.StatusCode < 200 || res.StatusCode > 299 { failed("status", res.StatusCode) return } attrs := []any{"kind", kind, "slug", slug, "status", res.StatusCode} var answer struct { Purged *int `json:"purged"` } if json.Unmarshal(body, &answer) == nil && answer.Purged != nil { attrs = append(attrs, "purged", *answer.Purged) } if coalesced > 0 { attrs = append(attrs, "coalesced", coalesced) } log.Info("cooler: site cache purged", attrs...) } // enqueue queues one purge. With nothing in flight it starts the worker, // which sends at once; while a request is in flight the body becomes the // pending follow-up, replacing (and counting) an earlier pending one. func (p *SiteCachePurger) enqueue(kind, slug string) { body := siteCachePurgeBody{Kind: kind, Slug: slug} p.mu.Lock() if p.sending { if p.pending != nil { p.coalesced++ } p.pending = &body p.mu.Unlock() return } p.sending = true p.idle = make(chan struct{}) p.mu.Unlock() go p.work(body) } // work sends body, then each pending follow-up, until none is left; then it // ends the busy period. func (p *SiteCachePurger) work(body siteCachePurgeBody) { coalesced := 0 for { p.send(context.Background(), body.Kind, body.Slug, coalesced) p.mu.Lock() if p.pending == nil { p.sending = false close(p.idle) p.idle = nil p.mu.Unlock() return } body, coalesced = *p.pending, p.coalesced p.pending, p.coalesced = nil, 0 p.mu.Unlock() } } // Wait blocks until no queued purge is pending or in flight, or until ctx // ends, then returning ctx.Err(). It returns nil at once for a nil or idle // purger. Tests and commands call it before they read what a purge did; // cooler's Handle.Close calls it at shutdown. func (p *SiteCachePurger) Wait(ctx context.Context) error { if p == nil { return nil } if ctx == nil { ctx = context.Background() } for { p.mu.Lock() idle := p.idle p.mu.Unlock() if idle == nil { return nil } select { case <-idle: // A new busy period may have started; look again. case <-ctx.Done(): return ctx.Err() } } } // purgeTransportError describes a failed request without the request URL // or headers. func purgeTransportError(err error) string { var uerr *url.Error if errors.As(err, &uerr) { err = uerr.Err } if errors.Is(err, context.DeadlineExceeded) { return "timeout" } return err.Error() } // PurgeSiteCacheAfterCommit queues a purge of kind and slug after the // transaction db belongs to commits, through lagoon.AfterCommit: a failed // save never purges, and a concurrent page view cannot re-cache the old // content between the purge and the commit. The after-commit callback only // queues, so the save never waits for the site (fire and forget). At most // one request is in flight per purger, and saves committed meanwhile // collapse into one follow-up request with the last kind and slug, since // each purge clears the site's whole route cache. A nil p does nothing; // outside a transaction (db nil or not in one) the purge is queued at once. // SiteCachePurger.Wait waits for the queue to drain. func PurgeSiteCacheAfterCommit(ctx context.Context, db *gorm.DB, p *SiteCachePurger, kind, slug string) { if p == nil { return } lagoon.AfterCommit(ctx, db, func(context.Context, *gorm.DB) { p.enqueue(kind, slug) }) } // SiteCacheFromApp returns the *SiteCachePurger published on app (by // Install, by a plugin's alias configuration or by an earlier call), or // builds one from site_cache.purge_url and site_cache.purge_token and // publishes it. It returns nil when neither key is set, and when the // configuration is invalid, which it logs as one error naming the key (the // serve command fails at boot on the same configuration). Admin hooks call // it, so it serves commands and tests that never run serve. func SiteCacheFromApp(app *backpack.App) *SiteCachePurger { if app == nil { return nil } if p, ok := app.Lookup[*SiteCachePurger](); ok && p != nil { return p } p, err := siteCacheFromConfig(app) if err != nil { loggerFromApp(app).Error("cooler: site cache purge disabled", "error", err.Error()) return nil } if p == nil { return nil } if err := app.Publish(p); err != nil { if winner, ok := app.Lookup[*SiteCachePurger](); ok && winner != nil { return winner } return nil } return p } // siteCacheFromConfig builds the purger from the framework keys. func siteCacheFromConfig(app *backpack.App) (*SiteCachePurger, error) { if app == nil || app.Config == nil { return nil, nil } p, err := NewSiteCachePurger(app.Config.String(SiteCachePurgeURLKey), app.Config.String(SiteCachePurgeTokenKey)) if err != nil { return nil, fmt.Errorf("cooler: %w", err) } return p, nil } // installSiteCache publishes the site-cache purger unless one is already // published (a plugin's alias configuration) and returns the purger in // effect, nil when none is configured: an invalid configuration is a boot // error naming the key. func installSiteCache(app *backpack.App, log *slog.Logger) (*SiteCachePurger, error) { if p, ok := app.Lookup[*SiteCachePurger](); ok && p != nil { log.Info("cooler: site cache purge enabled", "endpoint", p.Endpoint()) return p, nil } p, err := siteCacheFromConfig(app) if err != nil { return nil, err } if p == nil { return nil, nil } if err := app.Publish(p); err != nil { return nil, fmt.Errorf("cooler: %w", err) } log.Info("cooler: site cache purge enabled", "endpoint", p.Endpoint()) return p, nil }