monorepo/cloud/maplefile-backend/pkg/storage/database/cassandradb/cassandradb.go

// File Path: monorepo/cloud/maplefile-backend/pkg/storage/database/cassandradb/cassandradb.go
package cassandradb

import (
	"fmt"
	"strings"
	"time"

	"github.com/gocql/gocql"
	"go.uber.org/zap"

	"codeberg.org/mapleopentech/monorepo/cloud/maplefile-backend/config"
)

// CassandraDB wraps the gocql session with additional functionality
type CassandraDB struct {
	Session *gocql.Session
	config  config.DatabaseConfig
}

// gocqlLogger wraps zap logger to filter out noisy gocql warnings
type gocqlLogger struct {
	logger *zap.Logger
}

// Print implements gocql's Logger interface
func (l *gocqlLogger) Print(v ...interface{}) {
	msg := fmt.Sprint(v...)

	// Filter out noisy "invalid peer" warnings from Cassandra gossip
	// These are harmless and occur due to Docker networking
	if strings.Contains(msg, "Found invalid peer") {
		return
	}

	// Log other messages at debug level
	l.logger.Debug(msg)
}

// Printf implements gocql's Logger interface
func (l *gocqlLogger) Printf(format string, v ...interface{}) {
	msg := fmt.Sprintf(format, v...)

	// Filter out noisy "invalid peer" warnings from Cassandra gossip
	if strings.Contains(msg, "Found invalid peer") {
		return
	}

	// Log other messages at debug level
	l.logger.Debug(msg)
}

// Println implements gocql's Logger interface
func (l *gocqlLogger) Println(v ...interface{}) {
	msg := fmt.Sprintln(v...)

	// Filter out noisy "invalid peer" warnings from Cassandra gossip
	if strings.Contains(msg, "Found invalid peer") {
		return
	}

	// Log other messages at debug level
	l.logger.Debug(msg)
}

// NewCassandraConnection establishes a connection to Cassandra cluster
// Uses the simplified approach from MaplePress (working code)
func NewCassandraConnection(cfg *config.Config, logger *zap.Logger) (*gocql.Session, error) {
	dbConfig := cfg.Database

	logger.Info("⏳ Connecting to Cassandra...",
		zap.Strings("hosts", dbConfig.Hosts),
		zap.String("keyspace", dbConfig.Keyspace))

	// Create cluster configuration - let gocql handle DNS resolution
	cluster := gocql.NewCluster(dbConfig.Hosts...)
	cluster.Keyspace = dbConfig.Keyspace
	cluster.Consistency = parseConsistency(dbConfig.Consistency)
	cluster.ProtoVersion = 4
	cluster.ConnectTimeout = dbConfig.ConnectTimeout
	cluster.Timeout = dbConfig.RequestTimeout
	cluster.NumConns = 2

	// Set custom logger to filter out noisy warnings
	cluster.Logger = &gocqlLogger{logger: logger.Named("gocql")}

	// Retry policy
	cluster.RetryPolicy = &gocql.ExponentialBackoffRetryPolicy{
		NumRetries: int(dbConfig.MaxRetryAttempts),
		Min:        dbConfig.RetryDelay,
		Max:        10 * time.Second,
	}

	// Enable compression for better network efficiency
	cluster.Compressor = &gocql.SnappyCompressor{}

	// Create session
	session, err := cluster.CreateSession()
	if err != nil {
		return nil, fmt.Errorf("failed to connect to Cassandra: %w", err)
	}

	logger.Info("✓ Cassandra connected",
		zap.String("consistency", dbConfig.Consistency),
		zap.Int("connections", cluster.NumConns))

	return session, nil
}

// Close terminates the database connection
func (db *CassandraDB) Close() {
	if db.Session != nil {
		db.Session.Close()
	}
}

// Health checks if the database connection is still alive
func (db *CassandraDB) Health() error {
	// Quick health check using a simple query
	var timestamp time.Time
	err := db.Session.Query("SELECT now() FROM system.local").Scan(&timestamp)
	if err != nil {
		return fmt.Errorf("health check failed: %w", err)
	}

	// Validate that we got a reasonable timestamp (within last minute)
	now := time.Now()
	if timestamp.Before(now.Add(-time.Minute)) || timestamp.After(now.Add(time.Minute)) {
		return fmt.Errorf("health check returned suspicious timestamp: %v (current: %v)", timestamp, now)
	}

	return nil
}

// parseConsistency converts string consistency level to gocql.Consistency
func parseConsistency(consistency string) gocql.Consistency {
	switch consistency {
	case "ANY":
		return gocql.Any
	case "ONE":
		return gocql.One
	case "TWO":
		return gocql.Two
	case "THREE":
		return gocql.Three
	case "QUORUM":
		return gocql.Quorum
	case "ALL":
		return gocql.All
	case "LOCAL_QUORUM":
		return gocql.LocalQuorum
	case "EACH_QUORUM":
		return gocql.EachQuorum
	case "LOCAL_ONE":
		return gocql.LocalOne
	default:
		return gocql.Quorum // Default to QUORUM
	}
}