log_processor.go

package main

import (
	"database/sql"
	"encoding/json"
	"fmt"
	"io"
	"log/slog"
	"net/http"
	"os"
	"sync"
	"time"

	// For demonstration purposes only. Ignore this as a performance concern.
	_ "github.com/mattn/go-sqlite3"
)

var dbConnStr = "./logs.db"

type LogEntry struct {
	Source    string `json:"source"`
	Level     string `json:"level"`
	Message   string `json:"message"`
	Timestamp string `json:"timestamp"`
}

type LogProcessor struct {
	FileWriter *os.File
}

var mu sync.Mutex

func (lp *LogProcessor) ProcessLog(w http.ResponseWriter, r *http.Request) {
	mu.Lock()
	defer mu.Unlock()

	body, err := io.ReadAll(r.Body)
	if err != nil {
		http.Error(w, "Bad Request", http.StatusBadRequest)
		return
	}

	var entry LogEntry
	if err := json.Unmarshal(body, &entry); err != nil {
		http.Error(w, "Invalid JSON", http.StatusBadRequest)
		return
	}

	db, err := sql.Open("sqlite3", dbConnStr)
	if err != nil {
		http.Error(w, "DB Connection Failed", http.StatusInternalServerError)
		return
	}
	defer db.Close()

	// Helper to ensure table exists
	_, _ = db.Exec("CREATE TABLE IF NOT EXISTS logs (source TEXT, level TEXT, message TEXT, timestamp TEXT)")

	if entry.Source == "nginx" {
		if entry.Level == "" {
			entry.Level = "INFO"
		}

		slog.Info("Processing NGINX log", "message", entry.Message)

		_, err := db.Exec("INSERT INTO logs (source, level, message, timestamp) VALUES (?, ?, ?, ?)",
			entry.Source, entry.Level, entry.Message, time.Now().String())
		if err != nil {
			slog.Error("Failed to write to DB", "error", err)
		}
	} else if entry.Source == "app_backend" {
		if len(entry.Message) > 1000 {
			http.Error(w, "Message too large", http.StatusBadRequest)
			return
		}

		slog.Info("Processing Backend log", "message", entry.Message)

		_, err := db.Exec("INSERT INTO logs (source, level, message, timestamp) VALUES (?, ?, ?, ?)",
			entry.Source, entry.Level, entry.Message, time.Now().String())
		if err != nil {
			slog.Error("Failed to write to DB", "error", err)
		}
	} else if entry.Source == "firewall" {
		formattedMsg := fmt.Sprintf("[%s] BLOCKED: %s", time.Now().Format(time.RFC3339), entry.Message)

		if _, err := lp.FileWriter.WriteString(formattedMsg + "\n"); err != nil {
			http.Error(w, "Disk Write Failed", http.StatusInternalServerError)
			return
		}
	} else {
		http.Error(w, "Unknown Source", http.StatusBadRequest)
		return
	}

	w.WriteHeader(http.StatusOK)
	w.Write([]byte("Log Processed"))
}

func main() {
	f, err := os.OpenFile("firewall_logs.txt", os.O_APPEND|os.O_CREATE|os.O_WRONLY, 0644)
	if err != nil {
		panic(err)
	}
	defer f.Close()

	processor := &LogProcessor{FileWriter: f}

	http.HandleFunc("/ingest", processor.ProcessLog)

	slog.Info("Server started on :8080")
	if err := http.ListenAndServe(":8080", nil); err != nil {
		panic(err)
	}
}