84 lines
2.0 KiB
Go
84 lines
2.0 KiB
Go
package channelmgt
|
|
|
|
import (
|
|
"database/sql"
|
|
"encoding/json"
|
|
"fmt"
|
|
"os"
|
|
"time"
|
|
|
|
"github.com/lib/pq"
|
|
logging "lst.net/utils/logger"
|
|
)
|
|
|
|
// setup the notifiyer
|
|
|
|
// -- Only needs to be run once in DB
|
|
// CREATE OR REPLACE FUNCTION notify_new_log() RETURNS trigger AS $$
|
|
// BEGIN
|
|
// PERFORM pg_notify('new_log', row_to_json(NEW)::text);
|
|
// RETURN NEW;
|
|
// END;
|
|
// $$ LANGUAGE plpgsql;
|
|
|
|
// DROP TRIGGER IF EXISTS new_log_trigger ON logs;
|
|
|
|
// CREATE TRIGGER new_log_trigger
|
|
// AFTER INSERT ON logs
|
|
// FOR EACH ROW EXECUTE FUNCTION notify_new_log();
|
|
|
|
func AllLogs(db *sql.DB, broadcaster chan logging.Message) {
|
|
fmt.Println("[AllLogs] started")
|
|
log := logging.New()
|
|
dsn := fmt.Sprintf("host=%s port=%s user=%s password=%s dbname=%s sslmode=disable",
|
|
os.Getenv("DB_HOST"),
|
|
os.Getenv("DB_PORT"),
|
|
os.Getenv("DB_USER"),
|
|
os.Getenv("DB_PASSWORD"),
|
|
os.Getenv("DB_NAME"),
|
|
)
|
|
|
|
listener := pq.NewListener(dsn, 10*time.Second, time.Minute, nil)
|
|
err := listener.Listen("new_log")
|
|
if err != nil {
|
|
log.Panic("Failed to LISTEN on new_log", "logger", map[string]interface{}{
|
|
"error": err.Error(),
|
|
})
|
|
}
|
|
|
|
fmt.Println("Listening for new logs...")
|
|
|
|
for {
|
|
select {
|
|
case notify := <-listener.Notify:
|
|
if notify != nil {
|
|
fmt.Println("New log notification received")
|
|
|
|
// Unmarshal the JSON payload of the inserted row
|
|
var logData map[string]interface{}
|
|
if err := json.Unmarshal([]byte(notify.Extra), &logData); err != nil {
|
|
log.Error("Failed to unmarshal notification payload", "logger", map[string]interface{}{
|
|
"error": err.Error(),
|
|
})
|
|
continue
|
|
}
|
|
|
|
// Build message to broadcast
|
|
msg := logging.Message{
|
|
Channel: "logs", // This matches your logs channel name
|
|
Data: logData,
|
|
}
|
|
|
|
broadcaster <- msg
|
|
//fmt.Printf("[Broadcasting] sending: %+v\n", msg)
|
|
}
|
|
|
|
case <-time.After(90 * time.Second):
|
|
go func() {
|
|
log.Debug("Re-pinging Postgres LISTEN", "logger", map[string]interface{}{})
|
|
listener.Ping()
|
|
}()
|
|
}
|
|
}
|
|
}
|