// 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(×tamp) 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 } }