diff --git a/cmd/migrate/main.go b/cmd/migrate/main.go new file mode 100644 index 0000000..ad88942 --- /dev/null +++ b/cmd/migrate/main.go @@ -0,0 +1,290 @@ +package main + +import ( + "context" + "encoding/json" + "flag" + "fmt" + "log" + "time" + + "github.com/jackc/pgx/v5" + "github.com/jackc/pgx/v5/pgxpool" + "github.com/ostafen/clover/v2" + "github.com/ostafen/clover/v2/document" + "github.com/ostafen/clover/v2/query" + badgerstore "github.com/ostafen/clover/v2/store/badger" + + "pool-stats/database" + "pool-stats/models" +) + +const batchSize = 2000 + +var shareCopyColumns = []string{ + "work_info_id", "client_id", "enonce1", "nonce2", "nonce", "ntime", + "diff", "sdiff", "hash", "result", "errn", + "create_date", "create_ts", "create_by", "create_code", "create_inet", + "worker_name", "username", "address", "agent", +} + +func main() { + badgerPath := flag.String("BadgerPath", "badgerdb", "Path to the BadgerDB/Clover directory") + databaseDSN := flag.String("DatabaseDSN", + "postgres://admin:password@127.0.0.1:5432/pkstats?sslmode=disable", + "PostgreSQL connection string") + force := flag.Bool("Force", false, "Truncate existing Postgres tables before migrating") + flag.Parse() + + ctx := context.Background() + + log.Printf("Opening BadgerDB at %s...", *badgerPath) + store, err := badgerstore.Open(*badgerPath) + if err != nil { + log.Fatalf("failed to open BadgerDB: %v", err) + } + cloverDB, err := clover.OpenWithStore(store) + if err != nil { + log.Fatalf("failed to open CloverDB: %v", err) + } + defer cloverDB.Close() + + log.Printf("Connecting to PostgreSQL...") + pool, err := pgxpool.New(ctx, *databaseDSN) + if err != nil { + log.Fatalf("failed to connect to PostgreSQL: %v", err) + } + defer pool.Close() + + if err := database.ApplySchema(pool); err != nil { + log.Fatalf("failed to apply schema: %v", err) + } + + count, err := database.CountShares(pool) + if err != nil { + log.Fatalf("failed to count shares: %v", err) + } + if count > 0 { + if !*force { + log.Fatalf("refusing to migrate: shares table already has %d rows (use -Force to truncate and restart)", count) + } + log.Printf("Truncating existing tables (-Force)...") + if err := truncateAll(ctx, pool); err != nil { + log.Fatalf("failed to truncate: %v", err) + } + } + + shareCount, err := migrateShares(ctx, cloverDB, pool) + if err != nil { + log.Fatalf("failed to migrate shares: %v", err) + } + log.Printf("Migrated %d shares", shareCount) + + topCount, err := migrateTopShares(ctx, cloverDB, pool) + if err != nil { + log.Fatalf("failed to migrate top shares: %v", err) + } + log.Printf("Migrated %d top shares", topCount) + + twCount, err := migrateTimeWindowHighShares(ctx, cloverDB, pool) + if err != nil { + log.Fatalf("failed to migrate time window high shares: %v", err) + } + log.Printf("Migrated %d time window high shares", twCount) + + dailyCount, err := migrateDailyStats(ctx, cloverDB, pool) + if err != nil { + log.Fatalf("failed to migrate daily stats: %v", err) + } + log.Printf("Migrated %d daily stats", dailyCount) + + log.Println("Migration complete") +} + +func truncateAll(ctx context.Context, pool *pgxpool.Pool) error { + _, err := pool.Exec(ctx, ` + TRUNCATE TABLE shares, top_shares, time_window_high_shares, daily_stats + RESTART IDENTITY`) + return err +} + +func migrateShares(ctx context.Context, cloverDB *clover.DB, pool *pgxpool.Pool) (int, error) { + expected, err := cloverDB.Count(query.NewQuery(database.CollectionName)) + if err != nil { + return 0, fmt.Errorf("count shares: %w", err) + } + log.Printf("Streaming %d shares from Badger...", expected) + + total := 0 + batch := make([][]any, 0, batchSize) + lastLog := time.Now() + var walkErr error + + flush := func() error { + if len(batch) == 0 { + return nil + } + _, err := pool.CopyFrom( + ctx, + pgx.Identifier{"shares"}, + shareCopyColumns, + pgx.CopyFromRows(batch), + ) + if err != nil { + return err + } + total += len(batch) + batch = batch[:0] + if time.Since(lastLog) >= 2*time.Second || total == expected { + log.Printf(" shares progress: %d / %d", total, expected) + lastLog = time.Now() + } + return nil + } + + err = cloverDB.ForEach(query.NewQuery(database.CollectionName), func(doc *document.Document) bool { + var share models.ShareLog + if uerr := doc.Unmarshal(&share); uerr != nil { + walkErr = fmt.Errorf("unmarshal share: %w", uerr) + return false + } + vals, verr := database.ShareRowValues(share) + if verr != nil { + walkErr = verr + return false + } + batch = append(batch, vals) + if len(batch) >= batchSize { + if ferr := flush(); ferr != nil { + walkErr = ferr + return false + } + } + return true + }) + if err != nil { + return total, err + } + if walkErr != nil { + return total, walkErr + } + if err := flush(); err != nil { + return total, err + } + return total, nil +} + +func migrateTopShares(ctx context.Context, cloverDB *clover.DB, pool *pgxpool.Pool) (int, error) { + total := 0 + batch := make([][]any, 0, batchSize) + var walkErr error + + err := cloverDB.ForEach(query.NewQuery(database.TopSharesCollectionName), func(doc *document.Document) bool { + var share models.ShareLog + if uerr := doc.Unmarshal(&share); uerr != nil { + walkErr = fmt.Errorf("unmarshal top share: %w", uerr) + return false + } + vals, verr := database.ShareRowValues(share) + if verr != nil { + walkErr = verr + return false + } + batch = append(batch, vals) + total++ + return true + }) + if err != nil { + return 0, err + } + if walkErr != nil { + return 0, walkErr + } + if len(batch) == 0 { + return 0, nil + } + + _, err = pool.CopyFrom( + ctx, + pgx.Identifier{"top_shares"}, + shareCopyColumns, + pgx.CopyFromRows(batch), + ) + return total, err +} + +func migrateTimeWindowHighShares(ctx context.Context, cloverDB *clover.DB, pool *pgxpool.Pool) (int, error) { + count := 0 + var walkErr error + err := cloverDB.ForEach(query.NewQuery(database.TimeWindowHighShareCollectionName), func(doc *document.Document) bool { + var tw models.TimeWindowHighShare + if uerr := doc.Unmarshal(&tw); uerr != nil { + walkErr = fmt.Errorf("unmarshal time window high share: %w", uerr) + return false + } + if serr := database.SetTimeWindowHighShare(pool, tw); serr != nil { + walkErr = serr + return false + } + count++ + return true + }) + if err != nil { + return count, err + } + return count, walkErr +} + +func migrateDailyStats(ctx context.Context, cloverDB *clover.DB, pool *pgxpool.Pool) (int, error) { + count := 0 + var walkErr error + err := cloverDB.ForEach(query.NewQuery(database.DailyStatsCollectionName), func(doc *document.Document) bool { + var stats models.DailyStats + if uerr := doc.Unmarshal(&stats); uerr != nil { + walkErr = fmt.Errorf("unmarshal daily stats: %w", uerr) + return false + } + + topShareJSON, jerr := json.Marshal(stats.TopShare) + if jerr != nil { + walkErr = jerr + return false + } + workersJSON, jerr := json.Marshal(stats.Workers) + if jerr != nil { + walkErr = jerr + return false + } + + var expires any + if expiresAt := doc.ExpiresAt(); expiresAt != nil { + expires = *expiresAt + } + + dateStr := stats.Date + if dateStr == "" { + dateStr = time.Unix(0, 0).UTC().Format(time.DateOnly) + } + + _, ierr := pool.Exec(ctx, ` + INSERT INTO daily_stats (date, share_count, top_share, pool_hashrate, workers, expires_at) + VALUES ($1::date, $2, $3, $4, $5, $6) + ON CONFLICT (date) DO UPDATE SET + share_count = EXCLUDED.share_count, + top_share = EXCLUDED.top_share, + pool_hashrate = EXCLUDED.pool_hashrate, + workers = EXCLUDED.workers, + expires_at = EXCLUDED.expires_at`, + dateStr, stats.ShareCount, topShareJSON, stats.PoolHashrate, workersJSON, expires) + if ierr != nil { + walkErr = fmt.Errorf("insert daily stats for %s: %w", dateStr, ierr) + return false + } + count++ + return true + }) + if err != nil { + return count, err + } + return count, walkErr +} diff --git a/config/config.go b/config/config.go index 1e018b0..ae51618 100644 --- a/config/config.go +++ b/config/config.go @@ -5,14 +5,16 @@ import "flag" type Config struct { Port int `json:"port"` LogPath string `json:"logPath"` - DatabasePath string `json:"databasePath"` + DatabaseDSN string `json:"databaseDsn"` AdminPassword string `json:"adminPassword"` } func ParseFlags() Config { port := flag.Int("Port", 8080, "Listen port") logPath := flag.String("LogPath", "logs", "Path to log files") - databasePath := flag.String("DatabasePath", "badgerdb", "Path to the database directory") + databaseDSN := flag.String("DatabaseDSN", + "postgres://admin:password@127.0.0.1:5432/pkstats?sslmode=disable", + "PostgreSQL connection string") adminPassword := flag.String("AdminPassword", "", "Admin password for the web interface, disabled if empty") flag.Parse() @@ -20,7 +22,7 @@ func ParseFlags() Config { return Config{ Port: *port, LogPath: *logPath, - DatabasePath: *databasePath, + DatabaseDSN: *databaseDSN, AdminPassword: *adminPassword, } } diff --git a/database/db.go b/database/db.go index a31c93e..7a8ed29 100644 --- a/database/db.go +++ b/database/db.go @@ -1,6 +1,9 @@ package database import ( + "context" + _ "embed" + "encoding/json" "fmt" "log" "pool-stats/helpers" @@ -8,10 +11,8 @@ import ( "sort" "time" - "github.com/ostafen/clover/v2" - "github.com/ostafen/clover/v2/document" - c "github.com/ostafen/clover/v2/query" - badgerstore "github.com/ostafen/clover/v2/store/badger" + "github.com/jackc/pgx/v5" + "github.com/jackc/pgx/v5/pgxpool" ) const ( @@ -21,258 +22,283 @@ const ( DailyStatsCollectionName = "DailyStats" ) -func InitDatabase(path string) (*clover.DB, error) { - store, err := badgerstore.Open(path) +//go:embed schema.sql +var schemaSQL string + +func InitDatabase(dsn string) (*pgxpool.Pool, error) { + ctx := context.Background() + + pool, err := pgxpool.New(ctx, dsn) if err != nil { - return nil, fmt.Errorf("failed to open BadgerDB store: %v", err) + return nil, fmt.Errorf("failed to connect to PostgreSQL: %w", err) } - db, err := clover.OpenWithStore(store) - if err != nil { - return nil, fmt.Errorf("failed to open CloverDB: %v", err) + if err := pool.Ping(ctx); err != nil { + pool.Close() + return nil, fmt.Errorf("failed to ping PostgreSQL: %w", err) } - // Ensure collection exists - hasCollection, err := db.HasCollection(CollectionName) - if err != nil { - return nil, fmt.Errorf("failed to check collection: %v", err) + if _, err := pool.Exec(ctx, schemaSQL); err != nil { + pool.Close() + return nil, fmt.Errorf("failed to apply schema: %w", err) } - if !hasCollection { - if err := db.CreateCollection(CollectionName); err != nil { - return nil, fmt.Errorf("failed to create collection: %v", err) - } - if err := db.CreateIndex(CollectionName, "CreateDate"); err != nil { - return nil, fmt.Errorf("failed to create index: %v", err) - } - } - - // Init TopShares collection - hasTopSharesCollection, err := db.HasCollection(TopSharesCollectionName) - if err != nil { - return nil, fmt.Errorf("failed to check TopShares collection: %v", err) - } - - if !hasTopSharesCollection { - if err := db.CreateCollection(TopSharesCollectionName); err != nil { - return nil, fmt.Errorf("failed to create TopShares collection: %v", err) - } - - if err := db.CreateIndex(TopSharesCollectionName, "CreateDate"); err != nil { - return nil, fmt.Errorf("failed to create index for TopShares: %v", err) - } - - if err := db.CreateIndex(TopSharesCollectionName, "SDiff"); err != nil { - return nil, fmt.Errorf("failed to create index for TopShares SDiff: %v", err) - } - } - - // Init TimeWindowHighShareStat collection - hasTimeWindowCollection, err := db.HasCollection(TimeWindowHighShareCollectionName) - if err != nil { - return nil, fmt.Errorf("failed to check TimeWindowHighShare collection: %v", err) - } - - if !hasTimeWindowCollection { - if err := db.CreateCollection(TimeWindowHighShareCollectionName); err != nil { - return nil, fmt.Errorf("failed to create TimeWindowHighShare collection: %v", err) - } - if err := db.CreateIndex(TimeWindowHighShareCollectionName, "TimeWindowID"); err != nil { - return nil, fmt.Errorf("failed to create index for TimeWindowHighShare: %v", err) - } - } - - // Init DailyStats collection - hasDailyStatsCollection, err := db.HasCollection(DailyStatsCollectionName) - if err != nil { - return nil, fmt.Errorf("failed to check DailyStats collection: %v", err) - } - if !hasDailyStatsCollection { - if err := db.CreateCollection(DailyStatsCollectionName); err != nil { - return nil, fmt.Errorf("failed to create DailyStats collection: %v", err) - } - if err := db.CreateIndex(DailyStatsCollectionName, "Date"); err != nil { - return nil, fmt.Errorf("failed to create index for DailyStats: %v", err) - } - } - - return db, nil + return pool, nil } -func GetHighestSharesInRange(db *clover.DB, collection string, since time.Time, count int) ([]models.ShareLog, error) { - // Convert `since` to the format in `createdate` - lower := since.Unix() - upper := time.Now().Unix() +func shareColumns() string { + return `work_info_id, client_id, enonce1, nonce2, nonce, ntime, + diff, sdiff, hash, result, errn, + create_date, create_ts, create_by, create_code, create_inet, + worker_name, username, address, agent` +} - // Filter by timestamp range - criteria := c.Field("CreateDate").GtEq(fmt.Sprint(lower)). - And(c.Field("CreateDate").LtEq(fmt.Sprint(upper))) - - // Query sorted by "sdiff" descending, limit 1 - results, err := db.FindAll(c.NewQuery(collection). - Where(criteria). - Sort(c.SortOption{Field: "SDiff", Direction: -1}). - Limit(count)) - - if err != nil || len(results) == 0 { - return nil, err - } +func scanShare(row pgx.Row) (models.ShareLog, error) { + var s models.ShareLog + var createTs time.Time + err := row.Scan( + &s.WorkInfoID, &s.ClientID, &s.Enonce1, &s.Nonce2, &s.Nonce, &s.NTime, + &s.Diff, &s.SDiff, &s.Hash, &s.Result, &s.Errn, + &s.CreateDate, &createTs, &s.CreateBy, &s.CreateCode, &s.CreateInet, + &s.WorkerName, &s.Username, &s.Address, &s.Agent, + ) + return s, err +} +func scanShares(rows pgx.Rows) ([]models.ShareLog, error) { + defer rows.Close() var shares []models.ShareLog - for _, doc := range results { + for rows.Next() { var s models.ShareLog - if err := doc.Unmarshal(&s); err != nil { + var createTs time.Time + if err := rows.Scan( + &s.WorkInfoID, &s.ClientID, &s.Enonce1, &s.Nonce2, &s.Nonce, &s.NTime, + &s.Diff, &s.SDiff, &s.Hash, &s.Result, &s.Errn, + &s.CreateDate, &createTs, &s.CreateBy, &s.CreateCode, &s.CreateInet, + &s.WorkerName, &s.Username, &s.Address, &s.Agent, + ); err != nil { return nil, err } shares = append(shares, s) } - - return shares, nil + return shares, rows.Err() } -func PrintAllHashes(db *clover.DB) { - docs, err := db.FindAll(c.NewQuery(CollectionName)) +func shareValues(s models.ShareLog) ([]any, error) { + createTs, err := s.ParseCreateDate() if err != nil { - log.Fatalf("Failed to read from collection: %v", err) + createTs = helpers.ParseCreateDate(s.CreateDate) } + if createTs.IsZero() { + createTs = time.Unix(0, 0) + } + return []any{ + s.WorkInfoID, s.ClientID, s.Enonce1, s.Nonce2, s.Nonce, s.NTime, + s.Diff, s.SDiff, s.Hash, s.Result, s.Errn, + s.CreateDate, createTs, s.CreateBy, s.CreateCode, s.CreateInet, + s.WorkerName, s.Username, s.Address, s.Agent, + }, nil +} - for _, doc := range docs { - hash := doc.Get("Hash") - fmt.Println(hash) +func tableForCollection(collection string) string { + switch collection { + case TopSharesCollectionName: + return "top_shares" + default: + return "shares" } } -func ListShares(db *clover.DB, offset int, count int) []models.ShareLog { - results, err := db.FindAll( - c.NewQuery(CollectionName). - Sort(c.SortOption{Field: "CreateDate", Direction: -1}). - Skip(offset). - Limit(count), - ) +func InsertShare(db *pgxpool.Pool, share models.ShareLog) error { + ctx := context.Background() + vals, err := shareValues(share) + if err != nil { + return err + } + _, err = db.Exec(ctx, ` + INSERT INTO shares (`+shareColumns()+`) + VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12,$13,$14,$15,$16,$17,$18,$19,$20)`, + vals...) + return err +} + +func GetHighestSharesInRange(db *pgxpool.Pool, collection string, since time.Time, count int) ([]models.ShareLog, error) { + ctx := context.Background() + table := tableForCollection(collection) + rows, err := db.Query(ctx, ` + SELECT `+shareColumns()+` + FROM `+table+` + WHERE create_ts >= $1 AND create_ts <= $2 + ORDER BY sdiff DESC + LIMIT $3`, + since, time.Now(), count) + if err != nil { + return nil, err + } + return scanShares(rows) +} + +func ListShares(db *pgxpool.Pool, offset int, count int) []models.ShareLog { + ctx := context.Background() + rows, err := db.Query(ctx, ` + SELECT `+shareColumns()+` + FROM shares + ORDER BY create_ts DESC + OFFSET $1 LIMIT $2`, + offset, count) if err != nil { log.Printf("failed to list shares: %v", err) return nil } - - shareLogs := make([]models.ShareLog, len(results)) - for idx, doc := range results { - var shareLog models.ShareLog - doc.Unmarshal(&shareLog) - shareLogs[idx] = shareLog + shares, err := scanShares(rows) + if err != nil { + log.Printf("failed to scan shares: %v", err) + return nil } - - return shareLogs + return shares } -func ListTopShares(db *clover.DB) []models.ShareLog { - results, err := db.FindAll( - c.NewQuery(TopSharesCollectionName). - Sort(c.SortOption{Field: "SDiff", Direction: -1}), - ) +func ListTopShares(db *pgxpool.Pool) []models.ShareLog { + ctx := context.Background() + rows, err := db.Query(ctx, ` + SELECT `+shareColumns()+` + FROM top_shares + ORDER BY sdiff DESC`) if err != nil { log.Printf("failed to list top shares: %v", err) return nil } - - topShares := make([]models.ShareLog, len(results)) - for idx, doc := range results { - var shareLog models.ShareLog - doc.Unmarshal(&shareLog) - topShares[idx] = shareLog + shares, err := scanShares(rows) + if err != nil { + log.Printf("failed to scan top shares: %v", err) + return nil } - - return topShares + return shares } -func ReplaceTopShares(db *clover.DB, shares []models.ShareLog) { - db.Delete(c.NewQuery(TopSharesCollectionName)) +func ReplaceTopShares(db *pgxpool.Pool, shares []models.ShareLog) { + ctx := context.Background() + tx, err := db.Begin(ctx) + if err != nil { + log.Printf("failed to begin replace top shares: %v", err) + return + } + defer tx.Rollback(ctx) + + if _, err := tx.Exec(ctx, `DELETE FROM top_shares`); err != nil { + log.Printf("failed to clear top shares: %v", err) + return + } for _, share := range shares { - doc := document.NewDocumentOf(&share) - if _, err := db.InsertOne(TopSharesCollectionName, doc); err != nil { + vals, err := shareValues(share) + if err != nil { + log.Printf("failed to prepare top share: %v", err) + return + } + if _, err := tx.Exec(ctx, ` + INSERT INTO top_shares (`+shareColumns()+`) + VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12,$13,$14,$15,$16,$17,$18,$19,$20)`, + vals...); err != nil { + log.Printf("failed to insert top share: %v", err) return } } + + if err := tx.Commit(ctx); err != nil { + log.Printf("failed to commit top shares: %v", err) + } } -func GetTimeWindowHighShares(db *clover.DB) []models.TimeWindowHighShare { - results, err := db.FindAll( - c.NewQuery(TimeWindowHighShareCollectionName). - Sort(c.SortOption{Field: "TimeWindowID", Direction: 1}), - ) +func GetTimeWindowHighShares(db *pgxpool.Pool) []models.TimeWindowHighShare { + ctx := context.Background() + rows, err := db.Query(ctx, ` + SELECT time_window_id, time_window_name, sdiff, time + FROM time_window_high_shares + ORDER BY time_window_id ASC`) if err != nil { log.Printf("failed to list time window high shares: %v", err) return nil } + defer rows.Close() - timeWindowHighShares := make([]models.TimeWindowHighShare, len(results)) - for idx, doc := range results { - var timeWindowHighShare models.TimeWindowHighShare - doc.Unmarshal(&timeWindowHighShare) - timeWindowHighShares[idx] = timeWindowHighShare + var result []models.TimeWindowHighShare + for rows.Next() { + var tw models.TimeWindowHighShare + if err := rows.Scan(&tw.TimeWindowID, &tw.TimeWindowName, &tw.SDiff, &tw.Time); err != nil { + log.Printf("failed to scan time window high share: %v", err) + return nil + } + result = append(result, tw) } - - return timeWindowHighShares + return result } -func SetTimeWindowHighShare(db *clover.DB, share models.TimeWindowHighShare) error { - db.Delete( - c.NewQuery(TimeWindowHighShareCollectionName). - Where(c.Field("TimeWindowID"). - Eq(share.TimeWindowID))) - - doc := document.NewDocumentOf(&share) - db.InsertOne(TimeWindowHighShareCollectionName, doc) - - return nil +func SetTimeWindowHighShare(db *pgxpool.Pool, share models.TimeWindowHighShare) error { + ctx := context.Background() + _, err := db.Exec(ctx, ` + INSERT INTO time_window_high_shares (time_window_id, time_window_name, sdiff, time) + VALUES ($1, $2, $3, $4) + ON CONFLICT (time_window_id) DO UPDATE SET + time_window_name = EXCLUDED.time_window_name, + sdiff = EXCLUDED.sdiff, + time = EXCLUDED.time`, + share.TimeWindowID, share.TimeWindowName, share.SDiff, share.Time) + return err } -func ListSharesInTimeRange(db *clover.DB, since time.Time, till time.Time) []models.ShareLog { - lower := since.Unix() - upper := till.Unix() - - results, err := db.FindAll(c.NewQuery(CollectionName). - Where(c.Field("CreateDate").GtEq(fmt.Sprint(lower)). - And(c.Field("CreateDate").LtEq(fmt.Sprint(upper)))). - Sort(c.SortOption{Field: "CreateDate", Direction: -1})) - +func ListSharesInTimeRange(db *pgxpool.Pool, since time.Time, till time.Time) []models.ShareLog { + ctx := context.Background() + rows, err := db.Query(ctx, ` + SELECT `+shareColumns()+` + FROM shares + WHERE create_ts >= $1 AND create_ts <= $2 + ORDER BY create_ts DESC`, + since, till) if err != nil { log.Printf("failed to list shares in time range: %v", err) return nil } - - shareLogs := make([]models.ShareLog, len(results)) - for idx, doc := range results { - var shareLog models.ShareLog - doc.Unmarshal(&shareLog) - shareLogs[idx] = shareLog + shares, err := scanShares(rows) + if err != nil { + log.Printf("failed to scan shares in time range: %v", err) + return nil } - - return shareLogs + return shares } -// GetStatsForDay retrieves daily statistics for a given date -// Tries to find from DailyStats collection, calculates on the fly if not found and stores -func GetDailyStats(db *clover.DB, date time.Time) (*models.DailyStats, error) { +func GetDailyStats(db *pgxpool.Pool, date time.Time) (*models.DailyStats, error) { + ctx := context.Background() dateStr := date.Format(time.DateOnly) - // Check if stats already exist - existingDoc, err := db.FindFirst(c.NewQuery(DailyStatsCollectionName). - Where(c.Field("Date").Eq(dateStr))) - if err == nil && existingDoc != nil { - expiresAt := existingDoc.ExpiresAt() - if expiresAt != nil && expiresAt.After(time.Now()) { - DeleteDailyStatsForDay(db, date) - } else { - var stats models.DailyStats - if err := existingDoc.Unmarshal(&stats); err != nil { - return nil, fmt.Errorf("failed to unmarshal daily stats: %v", err) + var ( + stats models.DailyStats + topShare []byte + workers []byte + expiresAt *time.Time + ) + err := db.QueryRow(ctx, ` + SELECT date::text, share_count, top_share, pool_hashrate, workers, expires_at + FROM daily_stats + WHERE date = $1::date`, dateStr).Scan( + &stats.Date, &stats.ShareCount, &topShare, &stats.PoolHashrate, &workers, &expiresAt, + ) + if err == nil { + validCache := expiresAt == nil || expiresAt.After(time.Now()) + if validCache { + if err := json.Unmarshal(topShare, &stats.TopShare); err != nil { + return nil, fmt.Errorf("failed to unmarshal top share: %w", err) + } + if err := json.Unmarshal(workers, &stats.Workers); err != nil { + return nil, fmt.Errorf("failed to unmarshal workers: %w", err) } return &stats, nil } + _ = DeleteDailyStatsForDay(db, date) + } else if err != pgx.ErrNoRows { + return nil, fmt.Errorf("failed to query daily stats: %w", err) } - // Get shares in range since := date.Truncate(24 * time.Hour) till := since.Add(24 * time.Hour) shares := ListSharesInTimeRange(db, since, till) @@ -280,19 +306,17 @@ func GetDailyStats(db *clover.DB, date time.Time) (*models.DailyStats, error) { return shares[i].SDiff > shares[j].SDiff }) - // Calculate daily stats - stats := &models.DailyStats{ + computed := &models.DailyStats{ Date: dateStr, ShareCount: len(shares), Workers: make(map[string]models.WorkerDailyStats), } if len(shares) > 0 { - stats.TopShare = shares[0] - stats.PoolHashrate = helpers.CalculateAverageHashrate(shares) + computed.TopShare = shares[0] + computed.PoolHashrate = helpers.CalculateAverageHashrate(shares) } - // Calculate worker stats sharesByWorker := make(map[string][]models.ShareLog) for _, share := range shares { sharesByWorker[share.WorkerName] = append(sharesByWorker[share.WorkerName], share) @@ -302,46 +326,76 @@ func GetDailyStats(db *clover.DB, date time.Time) (*models.DailyStats, error) { sort.Slice(workerShares, func(i, j int) bool { return workerShares[i].SDiff > workerShares[j].SDiff }) - workerTopShare := workerShares[0] // Already sorted by SDiff - - stats.Workers[workerName] = models.WorkerDailyStats{ - TopShare: workerTopShare, + computed.Workers[workerName] = models.WorkerDailyStats{ + TopShare: workerShares[0], Hashrate: workerHashrate, Shares: len(workerShares), } } - // Insert or update the daily stats in the collection - doc := document.NewDocumentOf(stats) + topShareJSON, err := json.Marshal(computed.TopShare) + if err != nil { + return nil, fmt.Errorf("failed to marshal top share: %w", err) + } + workersJSON, err := json.Marshal(computed.Workers) + if err != nil { + return nil, fmt.Errorf("failed to marshal workers: %w", err) + } + var expires any isToday := dateStr == time.Now().UTC().Format(time.DateOnly) if isToday { - doc.SetExpiresAt(time.Now().Add(5 * time.Minute)) + expires = time.Now().Add(5 * time.Minute) } - if _, err := db.InsertOne(DailyStatsCollectionName, doc); err != nil { - return nil, fmt.Errorf("failed to insert daily stats: %v", err) + _, err = db.Exec(ctx, ` + INSERT INTO daily_stats (date, share_count, top_share, pool_hashrate, workers, expires_at) + VALUES ($1::date, $2, $3, $4, $5, $6) + ON CONFLICT (date) DO UPDATE SET + share_count = EXCLUDED.share_count, + top_share = EXCLUDED.top_share, + pool_hashrate = EXCLUDED.pool_hashrate, + workers = EXCLUDED.workers, + expires_at = EXCLUDED.expires_at`, + dateStr, computed.ShareCount, topShareJSON, computed.PoolHashrate, workersJSON, expires) + if err != nil { + return nil, fmt.Errorf("failed to insert daily stats: %w", err) } - return stats, nil + return computed, nil } -func ClearDailyStats(db *clover.DB) error { - // Delete all documents in DailyStats collection - if err := db.Delete(c.NewQuery(DailyStatsCollectionName)); err != nil { - return fmt.Errorf("failed to clear DailyStats collection: %v", err) +func ClearDailyStats(db *pgxpool.Pool) error { + ctx := context.Background() + if _, err := db.Exec(ctx, `DELETE FROM daily_stats`); err != nil { + return fmt.Errorf("failed to clear DailyStats: %w", err) } return nil } -func DeleteDailyStatsForDay(db *clover.DB, date time.Time) error { +func DeleteDailyStatsForDay(db *pgxpool.Pool, date time.Time) error { + ctx := context.Background() dateStr := date.Format(time.DateOnly) - - // Delete the document for the specific date - if err := db.Delete(c.NewQuery(DailyStatsCollectionName). - Where(c.Field("Date").Eq(dateStr))); err != nil { - return fmt.Errorf("failed to delete daily stats for %s: %v", dateStr, err) + if _, err := db.Exec(ctx, `DELETE FROM daily_stats WHERE date = $1::date`, dateStr); err != nil { + return fmt.Errorf("failed to delete daily stats for %s: %w", dateStr, err) } - return nil } + +// ShareRowValues returns ordered column values for COPY/batch insert. +func ShareRowValues(s models.ShareLog) ([]any, error) { + return shareValues(s) +} + +// ApplySchema runs the embedded schema against an existing pool. +func ApplySchema(db *pgxpool.Pool) error { + _, err := db.Exec(context.Background(), schemaSQL) + return err +} + +// CountShares returns the number of rows in shares. +func CountShares(db *pgxpool.Pool) (int64, error) { + var n int64 + err := db.QueryRow(context.Background(), `SELECT COUNT(*) FROM shares`).Scan(&n) + return n, err +} diff --git a/database/schema.sql b/database/schema.sql new file mode 100644 index 0000000..087242a --- /dev/null +++ b/database/schema.sql @@ -0,0 +1,69 @@ +CREATE TABLE IF NOT EXISTS shares ( + id BIGSERIAL PRIMARY KEY, + work_info_id BIGINT NOT NULL, + client_id INT NOT NULL, + enonce1 TEXT, + nonce2 TEXT, + nonce TEXT, + ntime TEXT, + diff DOUBLE PRECISION NOT NULL, + sdiff DOUBLE PRECISION NOT NULL, + hash TEXT, + result BOOLEAN, + errn INT, + create_date TEXT NOT NULL, + create_ts TIMESTAMPTZ NOT NULL, + create_by TEXT, + create_code TEXT, + create_inet TEXT, + worker_name TEXT, + username TEXT, + address TEXT, + agent TEXT +); + +CREATE INDEX IF NOT EXISTS shares_create_ts_desc ON shares (create_ts DESC); +CREATE INDEX IF NOT EXISTS shares_sdiff_desc ON shares (sdiff DESC); +CREATE INDEX IF NOT EXISTS shares_create_ts_sdiff ON shares (create_ts, sdiff DESC); + +CREATE TABLE IF NOT EXISTS top_shares ( + id BIGSERIAL PRIMARY KEY, + work_info_id BIGINT NOT NULL, + client_id INT NOT NULL, + enonce1 TEXT, + nonce2 TEXT, + nonce TEXT, + ntime TEXT, + diff DOUBLE PRECISION NOT NULL, + sdiff DOUBLE PRECISION NOT NULL, + hash TEXT, + result BOOLEAN, + errn INT, + create_date TEXT NOT NULL, + create_ts TIMESTAMPTZ NOT NULL, + create_by TEXT, + create_code TEXT, + create_inet TEXT, + worker_name TEXT, + username TEXT, + address TEXT, + agent TEXT +); + +CREATE INDEX IF NOT EXISTS top_shares_sdiff_desc ON top_shares (sdiff DESC); + +CREATE TABLE IF NOT EXISTS time_window_high_shares ( + time_window_id TEXT PRIMARY KEY, + time_window_name TEXT NOT NULL, + sdiff DOUBLE PRECISION NOT NULL, + time TEXT NOT NULL +); + +CREATE TABLE IF NOT EXISTS daily_stats ( + date DATE PRIMARY KEY, + share_count INT NOT NULL, + top_share JSONB NOT NULL, + pool_hashrate DOUBLE PRECISION NOT NULL, + workers JSONB NOT NULL, + expires_at TIMESTAMPTZ NULL +); diff --git a/go.mod b/go.mod index 2d949c0..46d9e07 100644 --- a/go.mod +++ b/go.mod @@ -1,9 +1,10 @@ module pool-stats -go 1.24.3 +go 1.25.0 require ( github.com/gofrs/uuid/v5 v5.3.1 + github.com/jackc/pgx/v5 v5.10.0 github.com/ostafen/clover/v2 v2.0.0-alpha.3.0.20250212110647-35f6fd38bde2 ) @@ -15,6 +16,9 @@ require ( github.com/golang/groupcache v0.0.0-20241129210726-2c02b8208cf8 // indirect github.com/google/flatbuffers v25.2.10+incompatible // indirect github.com/google/orderedcode v0.0.1 // indirect + github.com/jackc/pgpassfile v1.0.0 // indirect + github.com/jackc/pgservicefile v0.0.0-20240606120523-5a60cdf6a761 // indirect + github.com/jackc/puddle/v2 v2.2.2 // indirect github.com/klauspost/compress v1.18.0 // indirect github.com/pkg/errors v0.9.1 // indirect github.com/vmihailenco/msgpack/v5 v5.4.1 // indirect @@ -22,6 +26,8 @@ require ( go.etcd.io/bbolt v1.4.0 // indirect go.opencensus.io v0.24.0 // indirect golang.org/x/net v0.40.0 // indirect + golang.org/x/sync v0.17.0 // indirect golang.org/x/sys v0.33.0 // indirect + golang.org/x/text v0.29.0 // indirect google.golang.org/protobuf v1.36.6 // indirect ) diff --git a/go.sum b/go.sum index 5523ffd..bdd28af 100644 --- a/go.sum +++ b/go.sum @@ -51,6 +51,14 @@ github.com/google/go-cmp v0.6.0/go.mod h1:17dUlkBOakJ0+DkrSSNjCkIjxS6bF9zb3elmeN github.com/google/orderedcode v0.0.1 h1:UzfcAexk9Vhv8+9pNOgRu41f16lHq725vPwnSeiG/Us= github.com/google/orderedcode v0.0.1/go.mod h1:iVyU4/qPKHY5h/wSd6rZZCDcLJNxiWO6dvsYES2Sb20= github.com/google/uuid v1.1.2/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo= +github.com/jackc/pgpassfile v1.0.0 h1:/6Hmqy13Ss2zCq62VdNG8tM1wchn8zjSGOBJ6icpsIM= +github.com/jackc/pgpassfile v1.0.0/go.mod h1:CEx0iS5ambNFdcRtxPj5JhEz+xB6uRky5eyVu/W2HEg= +github.com/jackc/pgservicefile v0.0.0-20240606120523-5a60cdf6a761 h1:iCEnooe7UlwOQYpKFhBabPMi4aNAfoODPEFNiAnClxo= +github.com/jackc/pgservicefile v0.0.0-20240606120523-5a60cdf6a761/go.mod h1:5TJZWKEWniPve33vlWYSoGYefn3gLQRzjfDlhSJ9ZKM= +github.com/jackc/pgx/v5 v5.10.0 h1:VhSvgU2jSli8o3AqIEOTJr7rZwAEUVo4E4XhR94Zfr0= +github.com/jackc/pgx/v5 v5.10.0/go.mod h1:mal1tBGAFfLHvZzaYh77YS/eC6IX9OWbRV1QIIM0Jn4= +github.com/jackc/puddle/v2 v2.2.2 h1:PR8nw+E/1w0GLuRFSmiioY6UooMp6KJv0/61nB7icHo= +github.com/jackc/puddle/v2 v2.2.2/go.mod h1:vriiEXHvEE654aYKXXjOvZM39qJ0q+azkZFrfEOc3H4= github.com/klauspost/compress v1.18.0 h1:c/Cqfb0r+Yi+JtIEq73FWXVkRonBlf0CRNYc8Zttxdo= github.com/klauspost/compress v1.18.0/go.mod h1:2Pp+KzxcywXVXMr50+X0Q/Lsb43OQHYWRCY2AiWywWQ= github.com/ostafen/clover/v2 v2.0.0-alpha.3.0.20250212110647-35f6fd38bde2 h1:OzU5Rt6T/5Zba/mjRMUyTnAwRtfsRz8GLvJhtY65SU4= @@ -63,11 +71,13 @@ github.com/prometheus/client_model v0.0.0-20190812154241-14fe0d1b01d4/go.mod h1: github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME= github.com/stretchr/objx v0.4.0/go.mod h1:YvHI0jy2hoMjB+UWwv71VJQ9isScKT/TqJzVSSt89Yw= github.com/stretchr/objx v0.5.0/go.mod h1:Yh+to48EsGEfYuaHDzXPcE3xhTkx73EhmCGUpEOglKo= +github.com/stretchr/testify v1.3.0/go.mod h1:M5WIy9Dh21IEIfnGCwXGc5bZfKNJtfHm1UVUgZn+9EI= +github.com/stretchr/testify v1.7.0/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg= github.com/stretchr/testify v1.7.1/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg= github.com/stretchr/testify v1.8.0/go.mod h1:yNjHg4UonilssWZ8iaSj1OCr/vHnekPRkoO+kdMU+MU= github.com/stretchr/testify v1.8.1/go.mod h1:w2LPCIKwWwSfY2zedu0+kehJoqGctiVI29o6fzry7u4= -github.com/stretchr/testify v1.10.0 h1:Xv5erBjTwe/5IxqUQTdXv5kgmIvbHo3QQyRwhJsOfJA= -github.com/stretchr/testify v1.10.0/go.mod h1:r2ic/lqez/lEtzL7wO/rwa5dbSLXVDPFyf8C91i36aY= +github.com/stretchr/testify v1.11.1 h1:7s2iGBzp5EwR7/aIZr8ao5+dra3wiQyKjjFuvgVKu7U= +github.com/stretchr/testify v1.11.1/go.mod h1:wZwfW3scLgRK+23gO65QZefKpKQRnfz6sD981Nm4B6U= github.com/vmihailenco/msgpack/v5 v5.4.1 h1:cQriyiUvjTwOHg8QZaPihLWeRAAVoCpE00IUPn0Bjt8= github.com/vmihailenco/msgpack/v5 v5.4.1/go.mod h1:GaZTsDaehaPpQVyxrf5mtQlH+pc21PIudVV/E3rRQok= github.com/vmihailenco/tagparser/v2 v2.0.0 h1:y09buUbR+b5aycVFQs/g70pqKVZNBmxwAhO7/IwNM9g= @@ -94,8 +104,8 @@ golang.org/x/oauth2 v0.0.0-20180821212333-d2e6202438be/go.mod h1:N/0e6XlmueqKjAG golang.org/x/sync v0.0.0-20180314180146-1d60e4601c6f/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM= golang.org/x/sync v0.0.0-20181108010431-42b317875d0f/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM= golang.org/x/sync v0.0.0-20190423024810-112230192c58/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM= -golang.org/x/sync v0.11.0 h1:GGz8+XQP4FvTTrjZPzNKTMFtSXH80RAzG+5ghFPgK9w= -golang.org/x/sync v0.11.0/go.mod h1:Czt+wKu1gCyEFDUtn0jG5QVvpJ6rzVqr5aXyt9drQfk= +golang.org/x/sync v0.17.0 h1:l60nONMj9l5drqw6jlhIELNv9I0A4OFgRsG9k2oT9Ug= +golang.org/x/sync v0.17.0/go.mod h1:9KTHXmSnoGruLpwFjVSX0lNNA75CykiMECbovNTZqGI= golang.org/x/sys v0.0.0-20180830151530-49385e6e1522/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY= golang.org/x/sys v0.0.0-20190215142949-d0b11bdaac8a/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY= golang.org/x/sys v0.0.0-20190412213103-97732733099d/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs= @@ -104,6 +114,8 @@ golang.org/x/sys v0.33.0 h1:q3i8TbbEz+JRD9ywIRlyRAQbM0qF7hu24q3teo2hbuw= golang.org/x/sys v0.33.0/go.mod h1:BJP2sWEmIv4KK5OTEluFJCKSidICx8ciO85XgH3Ak8k= golang.org/x/text v0.3.0/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ= golang.org/x/text v0.3.3/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ= +golang.org/x/text v0.29.0 h1:1neNs90w9YzJ9BocxfsQNHKuAT4pkghyXc4nhZ6sJvk= +golang.org/x/text v0.29.0/go.mod h1:7MhJOA9CD2qZyOKYazxdYMF85OwPdEr9jTtBpO7ydH4= golang.org/x/tools v0.0.0-20180917221912-90fa682c2a6e/go.mod h1:n7NCudcB/nEzxVGmLbDWY5pfWTLqBcC2KZ6jyYvM4mQ= golang.org/x/tools v0.0.0-20190114222345-bf090417da8b/go.mod h1:n7NCudcB/nEzxVGmLbDWY5pfWTLqBcC2KZ6jyYvM4mQ= golang.org/x/tools v0.0.0-20190226205152-f727befe758c/go.mod h1:9Yl7xja0Znq3iFh3HoIrodX9oNMXvdceNzlUR8zjMvY= diff --git a/jobs/ingestSharesJob.go b/jobs/ingestSharesJob.go index dd871e4..3bda2b5 100644 --- a/jobs/ingestSharesJob.go +++ b/jobs/ingestSharesJob.go @@ -9,8 +9,7 @@ import ( "strings" "time" - "github.com/ostafen/clover/v2" - "github.com/ostafen/clover/v2/document" + "github.com/jackc/pgx/v5/pgxpool" "pool-stats/constants" "pool-stats/database" @@ -18,11 +17,11 @@ import ( ) type IngestSharesJob struct { - db *clover.DB + db *pgxpool.Pool logPath string } -func NewIngestSharesJob(db *clover.DB, path string) *IngestSharesJob { +func NewIngestSharesJob(db *pgxpool.Pool, path string) *IngestSharesJob { return &IngestSharesJob{db: db, logPath: path} } @@ -67,7 +66,7 @@ func (this *IngestSharesJob) ingestClosedBlocks() { } } -func (this *IngestSharesJob) ingestBlockDir(db *clover.DB, dirPath string) { +func (this *IngestSharesJob) ingestBlockDir(db *pgxpool.Pool, dirPath string) { files, err := os.ReadDir(dirPath) if err != nil { log.Printf("Failed to read block dir %s: %v", dirPath, err) @@ -97,8 +96,7 @@ func (this *IngestSharesJob) ingestBlockDir(db *clover.DB, dirPath string) { continue } - doc := document.NewDocumentOf(&share) - if _, err := db.InsertOne(database.CollectionName, doc); err != nil { + if err := database.InsertShare(db, share); err != nil { log.Println("DB insert error:", err) } } diff --git a/jobs/recalculateCurrentDayStatsJob.go b/jobs/recalculateCurrentDayStatsJob.go index 6977e15..66ed119 100644 --- a/jobs/recalculateCurrentDayStatsJob.go +++ b/jobs/recalculateCurrentDayStatsJob.go @@ -5,14 +5,14 @@ import ( "pool-stats/database" "time" - "github.com/ostafen/clover/v2" + "github.com/jackc/pgx/v5/pgxpool" ) type RecalculateCurrentDayStatsJob struct { - DB *clover.DB + DB *pgxpool.Pool } -func NewRecalculateCurrentDayStatsJob(db *clover.DB) *RecalculateCurrentDayStatsJob { +func NewRecalculateCurrentDayStatsJob(db *pgxpool.Pool) *RecalculateCurrentDayStatsJob { return &RecalculateCurrentDayStatsJob{DB: db} } diff --git a/jobs/recalculateTimeWindowHighShares.go b/jobs/recalculateTimeWindowHighShares.go index 30e1a1b..b838f8e 100644 --- a/jobs/recalculateTimeWindowHighShares.go +++ b/jobs/recalculateTimeWindowHighShares.go @@ -8,14 +8,14 @@ import ( "sort" "time" - "github.com/ostafen/clover/v2" + "github.com/jackc/pgx/v5/pgxpool" ) type RecalculateTimeWindowHighSharesJob struct { - DB *clover.DB + DB *pgxpool.Pool } -func NewRecalculateTimeWindowHighSharesJob(db *clover.DB) *RecalculateTimeWindowHighSharesJob { +func NewRecalculateTimeWindowHighSharesJob(db *pgxpool.Pool) *RecalculateTimeWindowHighSharesJob { return &RecalculateTimeWindowHighSharesJob{DB: db} } diff --git a/jobs/recalculateTopShares.go b/jobs/recalculateTopShares.go index 069c473..4b0e534 100644 --- a/jobs/recalculateTopShares.go +++ b/jobs/recalculateTopShares.go @@ -9,14 +9,14 @@ import ( "sort" "time" - "github.com/ostafen/clover/v2" + "github.com/jackc/pgx/v5/pgxpool" ) type RecalculateTopSharesJob struct { - DB *clover.DB + DB *pgxpool.Pool } -func NewRecalculateTopSharesJob(db *clover.DB) *RecalculateTopSharesJob { +func NewRecalculateTopSharesJob(db *pgxpool.Pool) *RecalculateTopSharesJob { return &RecalculateTopSharesJob{DB: db} } diff --git a/main.go b/main.go index 9733b76..728ff63 100644 --- a/main.go +++ b/main.go @@ -16,7 +16,7 @@ import ( func main() { config := config.ParseFlags() - db, err := database.InitDatabase(config.DatabasePath) + db, err := database.InitDatabase(config.DatabaseDSN) if err != nil { log.Fatalf("Failed to initialize database: %v", err) } diff --git a/web/server.go b/web/server.go index 507799f..b985bcc 100644 --- a/web/server.go +++ b/web/server.go @@ -9,18 +9,18 @@ import ( "fmt" "github.com/gofrs/uuid/v5" - "github.com/ostafen/clover/v2" + "github.com/jackc/pgx/v5/pgxpool" ) type WebServer struct { - db *clover.DB + db *pgxpool.Pool port int templates *template.Template sessions map[string]string adminPassword string } -func NewWebServer(db *clover.DB, port int, adminPassword string) *WebServer { +func NewWebServer(db *pgxpool.Pool, port int, adminPassword string) *WebServer { templates := template.New("base").Funcs(template.FuncMap{ "add": func(a, b int) int { return a + b }, "sub": func(a, b int) int { return a - b },