xt/app.go

package main

import (
	"context"
	"database/sql"
	"fmt"
	"io"
	"net/http"
	"os"
	"runtime/debug"
	"strconv"
	"strings"

	_ "embed"

	_ "github.com/mattn/go-sqlite3"
	"github.com/uptrace/opentelemetry-go-extra/otelsql"
	"go.opentelemetry.io/otel/attribute"
	semconv "go.opentelemetry.io/otel/semconv/v1.4.0"
	"go.opentelemetry.io/otel/trace"
	"go.sour.is/xt/internal/console"
	"go.sour.is/xt/internal/otel"
	"go.yarn.social/lextwt"
	"go.yarn.social/types"
	"golang.org/x/sync/errgroup"
)

func run(ctx context.Context, c *console.C[args]) error {
	ctx, span := otel.Span(ctx)
	defer span.End()

	bi, _ := debug.ReadBuildInfo()
	span.AddEvent(name, trace.WithAttributes(attribute.String("version", bi.Main.Version)))

	a := c.Args()
	app := &appState{
		args: a,
		C:    c,
		queue: FibHeap(func(a, b *Feed) bool {
			return a.NextScanOn.Time.Before(b.NextScanOn.Time)
		}),
	}

	// Setup DB
	err := func(ctx context.Context) error {
		ctx, span := otel.Span(ctx)
		defer span.End()

		db, err := app.DB(ctx)
		if err != nil {
			return err
		}
		defer db.Close()

		for _, stmt := range strings.Split(initSQL, ";") {
			_, err = db.ExecContext(ctx, stmt)
			if err != nil {
				return err
			}
		}

		return nil
	}(ctx)
	if err != nil {
		return err
	}

	// Seed File
	err = func(ctx context.Context) error {
		ctx, span := otel.Span(ctx)
		defer span.End()

		db, err := app.DB(ctx)
		if err != nil {
			return err
		}
		defer db.Close()

		var inFile io.Reader

		if a.baseFeed != "" {
			f, err := os.Open(a.baseFeed)
			if err != nil {
				return err
			}
			defer f.Close()
			err = storeRegistry(ctx, db, f)
			if err != nil {
				return err
			}
		}

		if a.URI != "" {
			res, err := http.Get(a.URI)
			if err != nil {
				return err
			}
			inFile = res.Body
			defer res.Body.Close()
			twtfile, err := lextwt.ParseFile(inFile, &types.Twter{
				Nick: a.Nick,
				URI:  a.URI,
			})
			if err != nil {
				return fmt.Errorf("%w: %w", ErrParseFailed, err)
			}

			return storeFeed(ctx, db, twtfile)
		}
		return nil
	}(ctx)
	if err != nil {
		return err
	}

	wg, ctx := errgroup.WithContext(ctx)

	wg.Go(func() error {
		return feedRefreshProcessor(ctx, app)
	})
	go httpServer(ctx, app)

	err = wg.Wait()
	if err != nil {
		return err
	}

	return ctx.Err()
}

type appState struct {
	args  args
	queue *fibHeap[Feed]
	*console.C[args]
}

type db struct {
	*sql.DB
	Version           string
	Params            map[string]string
	MaxLength         int
	MaxVariableNumber int
}

func (app *appState) DB(ctx context.Context) (db, error) {
	// return sql.Open(app.args.dbtype, app.args.dbfile)

	var err error
	db := db{Params: make(map[string]string)}
	db.DB, err = otelsql.Open(app.args.dbtype, app.args.dbfile,
		otelsql.WithAttributes(semconv.DBSystemSqlite),
		otelsql.WithDBName("xt"))
	if err != nil {
		return db, err
	}

	rows, err := db.DB.QueryContext(ctx, `select sqlite_version()`)
	if err != nil {
		return db, err
	}

	if rows.Next() {
		rows.Scan(&db.Version)
	}
	if err = rows.Err(); err != nil {
		return db, err
	}

	rows, err = db.DB.QueryContext(ctx, `pragma compile_options`)
	if err != nil {
		return db, err
	}

	for rows.Next() {
		var key, value string
		rows.Scan(&key)
		key, value, _ = strings.Cut(key, "=")
		db.Params[key] = value
	}
	if err = rows.Err(); err != nil {
		return db, err
	}

	if m, ok := db.Params["MAX_VARIABLE_NUMBER"]; ok {
		db.MaxVariableNumber, _ = strconv.Atoi(m)
	}
	if m, ok := db.Params["MAX_LENGTH"]; ok {
		db.MaxLength, _ = strconv.Atoi(m)
	}

	return db, err
}