OpenFlare/internal/model/openflare_access_log_store.go

// Copyright 2026 Arctel.net
// SPDX-License-Identifier: Apache-2.0

package model

import (
	"context"
	"sync"
	"time"

	"github.com/Rain-kl/Wavelet/internal/apps/openflare/chwriter"
	analyticsmodel "github.com/Rain-kl/Wavelet/internal/model/analytics"
	analyticsrepo "github.com/Rain-kl/Wavelet/internal/repository/analytics"
)

type accessLogStore interface {
	InsertBatch(ctx context.Context, records []*OpenFlareAccessLog) error
	List(ctx context.Context, query OpenFlareAccessLogQuery) ([]*OpenFlareAccessLog, error)
	Count(ctx context.Context, query OpenFlareAccessLogQuery) (int64, int64, error)
	RegionCounts(ctx context.Context, nodeID string, since time.Time, limit int) ([]*OpenFlareAccessLogRegionCount, error)
	BucketAggregates(ctx context.Context, filter OpenFlareAccessLogQuery, bucketSeconds int64) ([]openFlareAccessLogBucketAggregateRow, error)
	BucketDimensions(ctx context.Context, filter OpenFlareAccessLogQuery, column string, bucketSeconds int64) ([]openFlareAccessLogBucketDimensionRow, error)
	IPAggregates(ctx context.Context, filter OpenFlareAccessLogQuery, exactRemoteAddr bool) ([]openFlareAccessLogIPAggregateRow, error)
	IPSummaries(ctx context.Context, filter OpenFlareAccessLogQuery, recentSince time.Time) ([]openFlareAccessLogIPSummaryRow, error)
	IPTrend(ctx context.Context, filter OpenFlareAccessLogQuery, bucketSeconds int64) ([]openFlareAccessLogIPTrendRow, error)
	DeleteAll(ctx context.Context) (int64, error)
	DeleteBefore(ctx context.Context, cutoff time.Time) (int64, error)
	DeleteByNodeBefore(ctx context.Context, nodeID string, before time.Time) (int64, error)
}

var (
	accessLogStoreMu     sync.RWMutex
	accessLogStoreHolder accessLogStore
)

func currentAccessLogStore() accessLogStore {
	accessLogStoreMu.RLock()
	defer accessLogStoreMu.RUnlock()
	if accessLogStoreHolder != nil {
		return accessLogStoreHolder
	}
	return clickhouseAccessLogStore{}
}

// SetAccessLogStoreForTest swaps the access log store implementation for unit tests.
func SetAccessLogStoreForTest(store accessLogStore) func() {
	accessLogStoreMu.Lock()
	previous := accessLogStoreHolder
	accessLogStoreHolder = store
	accessLogStoreMu.Unlock()
	return func() {
		accessLogStoreMu.Lock()
		accessLogStoreHolder = previous
		accessLogStoreMu.Unlock()
	}
}

// NewMemoryAccessLogStore returns an in-memory access log store for unit tests.
func NewMemoryAccessLogStore() accessLogStore {
	return &memoryAccessLogStore{
		records: make([]*OpenFlareAccessLog, 0),
	}
}

type clickhouseAccessLogStore struct{}

func (clickhouseAccessLogStore) InsertBatch(_ context.Context, records []*OpenFlareAccessLog) error {
	logs := make([]analyticsmodel.NodeAccessLog, 0, len(records))
	for _, record := range records {
		if record == nil {
			continue
		}
		logs = append(logs, toAnalyticsNodeAccessLog(record))
	}
	chwriter.QueueNodeAccessLogs(logs)
	return nil
}

func (clickhouseAccessLogStore) List(ctx context.Context, query OpenFlareAccessLogQuery) ([]*OpenFlareAccessLog, error) {
	rows, err := analyticsrepo.ListNodeAccessLogs(ctx, toNodeAccessLogFilter(query))
	if err != nil {
		return nil, err
	}
	return fromAnalyticsNodeAccessLogs(rows), nil
}

func (clickhouseAccessLogStore) Count(ctx context.Context, query OpenFlareAccessLogQuery) (int64, int64, error) {
	return analyticsrepo.CountNodeAccessLogs(ctx, toNodeAccessLogFilter(query))
}

func (clickhouseAccessLogStore) RegionCounts(ctx context.Context, nodeID string, since time.Time, limit int) ([]*OpenFlareAccessLogRegionCount, error) {
	rows, err := analyticsrepo.RegionCountsNodeAccessLogs(ctx, nodeID, since, limit)
	if err != nil {
		return nil, err
	}
	result := make([]*OpenFlareAccessLogRegionCount, len(rows))
	for index, row := range rows {
		result[index] = &OpenFlareAccessLogRegionCount{
			Region: row.Region,
			Count:  row.Count,
		}
	}
	return result, nil
}

func (clickhouseAccessLogStore) BucketAggregates(ctx context.Context, filter OpenFlareAccessLogQuery, bucketSeconds int64) ([]openFlareAccessLogBucketAggregateRow, error) {
	rows, err := analyticsrepo.BucketAggregatesNodeAccessLogs(ctx, toNodeAccessLogFilter(filter), bucketSeconds)
	if err != nil {
		return nil, err
	}
	result := make([]openFlareAccessLogBucketAggregateRow, len(rows))
	for index, row := range rows {
		result[index] = openFlareAccessLogBucketAggregateRow{
			BucketEpoch:      row.BucketEpoch,
			RequestCount:     row.RequestCount,
			SuccessCount:     row.SuccessCount,
			ClientErrorCount: row.ClientErrorCount,
			ServerErrorCount: row.ServerErrorCount,
		}
	}
	return result, nil
}

func (clickhouseAccessLogStore) BucketDimensions(ctx context.Context, filter OpenFlareAccessLogQuery, column string, bucketSeconds int64) ([]openFlareAccessLogBucketDimensionRow, error) {
	rows, err := analyticsrepo.BucketDimensionsNodeAccessLogs(ctx, toNodeAccessLogFilter(filter), column, bucketSeconds)
	if err != nil {
		return nil, err
	}
	result := make([]openFlareAccessLogBucketDimensionRow, len(rows))
	for index, row := range rows {
		result[index] = openFlareAccessLogBucketDimensionRow{
			BucketEpoch: row.BucketEpoch,
			Value:       row.Value,
		}
	}
	return result, nil
}

func (clickhouseAccessLogStore) IPAggregates(ctx context.Context, filter OpenFlareAccessLogQuery, exactRemoteAddr bool) ([]openFlareAccessLogIPAggregateRow, error) {
	rows, err := analyticsrepo.IPAggregatesNodeAccessLogs(ctx, toNodeAccessLogFilter(filter), exactRemoteAddr)
	if err != nil {
		return nil, err
	}
	result := make([]openFlareAccessLogIPAggregateRow, len(rows))
	for index, row := range rows {
		result[index] = openFlareAccessLogIPAggregateRow{
			RemoteAddr:       row.RemoteAddr,
			RequestCount:     row.RequestCount,
			SuccessCount:     row.SuccessCount,
			ClientErrorCount: row.ClientErrorCount,
			ServerErrorCount: row.ServerErrorCount,
			LastSeenEpoch:    row.LastSeenEpoch,
		}
	}
	return result, nil
}

func (clickhouseAccessLogStore) IPSummaries(ctx context.Context, filter OpenFlareAccessLogQuery, recentSince time.Time) ([]openFlareAccessLogIPSummaryRow, error) {
	rows, err := analyticsrepo.IPSummariesNodeAccessLogs(ctx, toNodeAccessLogFilter(filter), recentSince)
	if err != nil {
		return nil, err
	}
	result := make([]openFlareAccessLogIPSummaryRow, len(rows))
	for index, row := range rows {
		result[index] = openFlareAccessLogIPSummaryRow{
			RemoteAddr:     row.RemoteAddr,
			TotalRequests:  row.TotalRequests,
			RecentRequests: row.RecentRequests,
			LastSeenEpoch:  row.LastSeenEpoch,
		}
	}
	return result, nil
}

func (clickhouseAccessLogStore) IPTrend(ctx context.Context, filter OpenFlareAccessLogQuery, bucketSeconds int64) ([]openFlareAccessLogIPTrendRow, error) {
	rows, err := analyticsrepo.IPTrendNodeAccessLogs(ctx, toNodeAccessLogFilter(filter), bucketSeconds)
	if err != nil {
		return nil, err
	}
	result := make([]openFlareAccessLogIPTrendRow, len(rows))
	for index, row := range rows {
		result[index] = openFlareAccessLogIPTrendRow{
			BucketEpoch:  row.BucketEpoch,
			RequestCount: row.RequestCount,
		}
	}
	return result, nil
}

func (clickhouseAccessLogStore) DeleteAll(ctx context.Context) (int64, error) {
	return analyticsrepo.DeleteAllNodeAccessLogs(ctx)
}

func (clickhouseAccessLogStore) DeleteBefore(ctx context.Context, cutoff time.Time) (int64, error) {
	return analyticsrepo.DeleteNodeAccessLogsBefore(ctx, cutoff)
}

func (clickhouseAccessLogStore) DeleteByNodeBefore(ctx context.Context, nodeID string, before time.Time) (int64, error) {
	return analyticsrepo.DeleteNodeAccessLogsByNodeBefore(ctx, nodeID, before)
}

func toNodeAccessLogFilter(query OpenFlareAccessLogQuery) analyticsrepo.NodeAccessLogFilter {
	return analyticsrepo.NodeAccessLogFilter{
		NodeID:     query.NodeID,
		RemoteAddr: query.RemoteAddr,
		Host:       query.Host,
		Path:       query.Path,
		Since:      query.Since,
		Until:      query.Until,
		Page:       query.Page,
		PageSize:   query.PageSize,
		SortBy:     query.SortBy,
		SortOrder:  query.SortOrder,
	}
}

func toAnalyticsNodeAccessLog(record *OpenFlareAccessLog) analyticsmodel.NodeAccessLog {
	return analyticsmodel.NodeAccessLog{
		ID:         record.ID,
		NodeID:     record.NodeID,
		LoggedAt:   record.LoggedAt,
		RemoteAddr: record.RemoteAddr,
		Region:     record.Region,
		Host:       record.Host,
		Path:       record.Path,
		StatusCode: openFlareAccessLogStatusCodeToInt32(record.StatusCode),
		CreatedAt:  record.CreatedAt,
	}
}

func fromAnalyticsNodeAccessLogs(rows []analyticsmodel.NodeAccessLog) []*OpenFlareAccessLog {
	result := make([]*OpenFlareAccessLog, len(rows))
	for index, row := range rows {
		result[index] = &OpenFlareAccessLog{
			ID:         row.ID,
			NodeID:     row.NodeID,
			LoggedAt:   row.LoggedAt,
			RemoteAddr: row.RemoteAddr,
			Region:     row.Region,
			Host:       row.Host,
			Path:       row.Path,
			StatusCode: int(row.StatusCode),
			CreatedAt:  row.CreatedAt,
		}
	}
	return result
}