module.go

package xrss

import (
	"context"
	"database/sql"
	"fmt"
	"io/fs"
	"log/slog"
	"net/http"
	"os"
	"path/filepath"
	"regexp"
	"runtime"
	"strconv"
	"time"

	"github.com/caddyserver/caddy/v2"
	"github.com/caddyserver/caddy/v2/caddyconfig/httpcaddyfile"
	"github.com/caddyserver/caddy/v2/modules/caddyhttp"
	"github.com/infogulch/watch"
	"github.com/infogulch/xtemplate"
	"github.com/infogulch/xtemplate/providers"
	natsxt "github.com/infogulch/xtemplate/providers/nats"
	"github.com/nats-io/nats-server/v2/server"
	"github.com/nats-io/nats.go"
	"github.com/nats-io/nats.go/jetstream"
	"go.uber.org/zap/exp/zapslog"

	_ "github.com/mattn/go-sqlite3"
)

func init() {
	caddy.RegisterModule(XRssModule{})
	httpcaddyfile.RegisterHandlerDirective("xrss", func(h httpcaddyfile.Helper) (caddyhttp.MiddlewareHandler, error) {
		return &XRssModule{}, nil
	})
}

// CaddyModule returns the Caddy module information.
func (XRssModule) CaddyModule() caddy.ModuleInfo {
	return caddy.ModuleInfo{
		ID:  "http.handlers.xrss",
		New: func() caddy.Module { return new(XRssModule) },
	}
}

func (m *XRssModule) ServeHTTP(w http.ResponseWriter, r *http.Request, _ caddyhttp.Handler) error {
	m.Handler.ServeHTTP(w, r)
	return nil
}

var _ caddyhttp.MiddlewareHandler = (*XRssModule)(nil)

type XRssModule struct {
	http.Handler
	cancel func()
}

// Provision provisions t. Implements caddy.Provisioner.
func (m *XRssModule) Provision(ctx caddy.Context) error {
	srv, err := server.NewServer(&server.Options{DontListen: true, JetStream: true})
	if err != nil {
		return fmt.Errorf("failed to create nats server: %w", err)
	}
	srv.Start()

	conn, err := nats.Connect("", nats.InProcessServer(srv))
	if err != nil {
		return fmt.Errorf("failed to create nats conn: %w", err)
	}

	js, err := jetstream.New(conn)
	if err != nil {
		return fmt.Errorf("failed to create jetstream: %w", err)
	}

	kv, err := js.CreateKeyValue(context.Background(), jetstream.KeyValueConfig{Bucket: "rss"})
	if err != nil {
		return fmt.Errorf("failed to create key value: %w", err)
	}

	connstr := "file:rss.sqlite?_journal=WAL&_synchronous=NORMAL&cache=shared&_cache_size=1000000000&_busy_timeout=5000"

	wdb, err := sql.Open("sqlite3", connstr+"&mode=rwc&_txlock=immediate&_foreign_keys=true&_vacuum=full")
	if err != nil {
		return fmt.Errorf("failed to create writable db: %w", err)
	}
	wdb.SetMaxOpenConns(1)

	rdb, err := sql.Open("sqlite3", connstr+"&mode=ro&_txlock=deferred")
	if err != nil {
		return fmt.Errorf("failed to created readonly db: %w", err)
	}
	rdb.SetMaxOpenConns(runtime.NumCPU())

	// Wrap zap logger into a slog logger for xtemplate
	log := slog.New(zapslog.NewHandler(ctx.Logger().Core(), nil)).WithGroup("xrss")

	err = migrate(wdb, log)
	if err != nil {
		return fmt.Errorf("failed to apply migrations: %w", err)
	}

	for i := range 5 {
		wlog := log.With(slog.Int("fetch-worker", i))
		go fetchWorker(wdb, rdb, kv, conn, wlog)
	}

	config, err := xtemplate.New().Defaults().Options(
		providers.WithKV("Config", make(map[string]string)),
		providers.WithDB("DB", rdb, &sql.TxOptions{ReadOnly: true}),
		natsxt.WithKV("KV", kv),
		natsxt.WithConn("Nats", conn),
		xtemplate.WithLogger(log),
	)
	if err != nil {
		return fmt.Errorf("failed to configure xtemplate: %w", err)
	}

	server, err := config.Server()
	if err != nil {
		return fmt.Errorf("failed to start xtemplate server: %w", err)
	}

	halt, err := watch.Watch([]string{"templates"}, 200*time.Millisecond, log.WithGroup("fswatch"), func() bool {
		server.Reload()
		return true
	})
	if err != nil {
		return fmt.Errorf("failed to start fswatch: %w", err)
	}

	m.Handler = server.Handler()

	m.cancel = func() {
		server.Stop()
		srv.Shutdown()
		conn.Close()
		wdb.Close()
		rdb.Close()
		halt <- struct{}{}
		m.Handler = nil
		m.cancel = nil
	}

	return nil
}

// Cleanup discards resources held by t. Implements caddy.CleanerUpper.
func (m *XRssModule) Cleanup() error {
	if m.cancel != nil {
		m.cancel()
	}
	return nil
}

func migrate(db *sql.DB, log *slog.Logger) error {
	var start_user_version, curent_user_version int64
	err := db.QueryRow(`PRAGMA user_version;`).Scan(&start_user_version)
	if err != nil {
		return fmt.Errorf("failed to get current user_version: %w", err)
	}

	count := 0
	re := regexp.MustCompile(`schema.(\d+).sql`)
	if err = filepath.WalkDir("./migrations", func(path string, d fs.DirEntry, err error) error {
		if err != nil || d.IsDir() {
			return err
		}
		m := re.FindStringSubmatch(filepath.Base(path))
		if m == nil {
			return fmt.Errorf("failed to match schema migration file '%s'", path)
		}
		i, _ := strconv.ParseInt(m[1], 10, 0)
		if i <= start_user_version {
			log.Debug("skipped file", slog.String("path", path), slog.Int64("path_version", i), slog.Int64("start_user_version", start_user_version))
			return err
		}
		log.Debug("applying migration", slog.String("path", path), slog.Int64("path_version", i), slog.Int64("start_user_version", start_user_version))
		migration, err := os.ReadFile(path)
		if err != nil {
			return fmt.Errorf("failed to read migration file '%s': %w", path, err)
		}
		_, err = db.Exec(string(migration))
		if err != nil {
			return fmt.Errorf("failed to exec migration file '%s': %w", path, err)
		}
		err = db.QueryRow(`PRAGMA user_version;`).Scan(&curent_user_version)
		if err != nil {
			return fmt.Errorf("failed to get current user_version: %w", err)
		}
		log.Debug("applied migration", slog.String("file", path), slog.Int64("curent_user_version", curent_user_version))
		count++
		return nil
	}); err != nil {
		return fmt.Errorf("failed to apply migration file: %w", err)
	}

	var end_user_version int64
	err = db.QueryRow(`PRAGMA user_version;`).Scan(&end_user_version)
	if err != nil {
		return fmt.Errorf("failed to get current user_version: %w", err)
	}

	log.Info("applied migrations", slog.Int("count", count), slog.Int64("start_user_version", start_user_version), slog.Int64("curent_user_version", curent_user_version))
	return nil
}