home-automation-backend/src/components/pooRecorder/pooRecorder.go

package pooRecorder

import (
	"database/sql"
	"encoding/json"
	"fmt"
	"net/http"
	"os"
	"strconv"
	"strings"
	"time"

	"log/slog"

	"github.com/go-co-op/gocron/v2"
	"github.com/jomei/notionapi"
	"github.com/spf13/viper"
	"github.com/t-liu93/home-automation-backend/util/homeassistantutil"
	"github.com/t-liu93/home-automation-backend/util/notion"
	_ "modernc.org/sqlite"
)

var (
	db        *sql.DB
	scheduler *gocron.Scheduler
)

type recordDetail struct {
	Status    string `json:"status"`
	Latitude  string `json:"latitude"`
	Longitude string `json:"longitude"`
}

type pooStatusSensorAttributes struct {
	LastPoo string `json:"last_poo"`
}

type pooStatusWebhookBody struct {
	Status string `json:"status"`
}

type pooStatusDbEntry struct {
	Timestamp string
	Status    string
	Latitude  float64
	Longitude float64
}

func Init(mainScheduler *gocron.Scheduler) {
	initDb()
	initScheduler(mainScheduler)
}

func HandleRecordPoo(w http.ResponseWriter, r *http.Request) {
	var record recordDetail
	if !viper.InConfig("pooRecorder.tableId") {
		slog.Warn("HandleRecordPoo Table ID not found in config file")
		http.Error(w, "Table ID not found in config file", http.StatusInternalServerError)
		return
	}
	decoder := json.NewDecoder(r.Body)
	decoder.DisallowUnknownFields()
	err := decoder.Decode(&record)
	if err != nil {
		slog.Warn(fmt.Sprintln("HandleRecordPoo Error decoding request body", err))
		http.Error(w, err.Error(), http.StatusBadRequest)
		return
	}
	now := time.Now()
	err = storeStatus(record, now)
	if err != nil {
		slog.Warn(fmt.Sprintln("HandleRecordPoo Error storing status", err))
		http.Error(w, err.Error(), http.StatusInternalServerError)
		return
	}
	pooStatus := homeassistantutil.HttpSensor{
		EntityId: "sensor.test_poo_sensor",
		State:    record.Status,
		Attributes: pooStatusSensorAttributes{
			LastPoo: now.Format("Mon | 2006-01-02 | 15:04"),
		},
	}
	homeassistantutil.PublishSensor(pooStatus)
	if viper.InConfig("pooRecorder.webhookId") {
		homeassistantutil.TriggerWebhook(viper.GetString("pooRecorder.webhookId"), pooStatusWebhookBody{Status: record.Status})
	} else {
		slog.Warn("HandleRecordPoo Webhook ID not found in config file")
	}
}

func HandleNotifyLatestPoo(w http.ResponseWriter, r *http.Request) {
	var latest pooStatusDbEntry
	err := db.QueryRow(`SELECT timestamp, status, latitude, longitude FROM poo_records ORDER BY timestamp DESC LIMIT 1`).Scan(&latest.Timestamp, &latest.Status, &latest.Latitude, &latest.Longitude)
	if err != nil {
		slog.Warn(fmt.Sprintln("HandleGetLatestPoo Error getting latest poo", err))
		http.Error(w, err.Error(), http.StatusInternalServerError)
		return
	}
	recordTime, err := time.Parse("2006-01-02T15:04Z07:00", latest.Timestamp)
	if err != nil {
		slog.Warn(fmt.Sprintln("HandleGetLatestPoo Error parsing timestamp", err))
		http.Error(w, err.Error(), http.StatusInternalServerError)
		return
	}
	recordTime = recordTime.Local()
	pooStatus := homeassistantutil.HttpSensor{
		EntityId: "sensor.test_poo_sensor",
		State:    latest.Status,
		Attributes: pooStatusSensorAttributes{
			LastPoo: recordTime.Format("Mon | 2006-01-02 | 15:04"),
		},
	}
	homeassistantutil.PublishSensor(pooStatus)
	slog.Debug(fmt.Sprintln("HandleGetLatestPoo Latest poo", pooStatus.State, "at", pooStatus.Attributes.(pooStatusSensorAttributes).LastPoo))
}

func initDb() {
	if !viper.InConfig("pooRecorder.dbPath") {
		slog.Info("HandleRecordPoo dbPath not found in config file, using default: pooRecorder.db")
		viper.SetDefault("pooRecorder.dbPath", "pooRecorder.db")
	}

	dbPath := viper.GetString("pooRecorder.dbPath")
	err := error(nil)
	db, err = sql.Open("sqlite", dbPath)
	if err != nil {
		slog.Error(fmt.Sprintln("PooRecorderInit Error opening database", err))
		os.Exit(1)
	}
	err = db.Ping()
	if err != nil {
		slog.Error(fmt.Sprintln("PooRecorderInit Error pinging database", err))
		os.Exit(1)
	}
	migrateDb()
}

func migrateDb() {
	var userVersion int
	err := db.QueryRow("PRAGMA user_version").Scan(&userVersion)
	if err != nil {
		slog.Error(fmt.Sprintln("PooRecorderInit Error getting db user version", err))
		os.Exit(1)
	}
	if userVersion == 0 {
		migrateDb0To1(&userVersion)
	}
}

func migrateDb0To1(userVersion *int) {
	// this is actually create new db
	slog.Info("Creating database version 1..")
	_, err := db.Exec(`CREATE TABLE IF NOT EXISTS poo_records (
	timestamp TEXT NOT NULL,
	status TEXT NOT NULL,
	latitude REAL NOT NULL,
	longitude REAL NOT NULL,
	PRIMARY KEY (timestamp))`)
	if err != nil {
		slog.Error(fmt.Sprintln("PooRecorderInit Error creating table", err))
		os.Exit(1)
	}
	_, err = db.Exec(`PRAGMA user_version = 1`)
	if err != nil {
		slog.Error(fmt.Sprintln("PooRecorderInit Error setting user version to 1", err))
		os.Exit(1)
	}
	*userVersion = 1
}

func initScheduler(mainScheduler *gocron.Scheduler) {
	scheduler = mainScheduler
	_, err := (*scheduler).NewJob(gocron.CronJob("0 5 * * *", false), gocron.NewTask(
		notionDbSync,
	))
	if err != nil {
		slog.Error(fmt.Sprintln("PooRecorderInit Error creating scheduled task", err))
		os.Exit(1)
	}
}

func notionDbSync() {
	slog.Info("PooRecorder Running DB sync with Notion..")
	if !viper.InConfig("pooRecorder.tableId") {
		slog.Warn("PooRecorder Table ID not found in config file, sync aborted")
		return
	}
	tableId := viper.GetString("pooRecorder.tableId")
	rowsNotion, err := notion.GetAllTableRows(tableId)
	if err != nil {
		slog.Error(fmt.Sprintln("PooRecorderSyncDb Failed to get table header", err))
		return
	}
	header := rowsNotion[0]
	rowsNotion = rowsNotion[1:] // remove header
	rowsDb, err := db.Query(`SELECT * FROM poo_records`)
	rowsDbMap := make(map[string]pooStatusDbEntry)
	if err != nil {
		slog.Error(fmt.Sprintln("PooRecorderSyncDb Failed to get db rows", err))
		return
	}
	defer rowsDb.Close()
	for rowsDb.Next() {
		var row pooStatusDbEntry
		err = rowsDb.Scan(&row.Timestamp, &row.Status, &row.Latitude, &row.Longitude)
		if err != nil {
			slog.Error(fmt.Sprintln("PooRecorderSyncDb Failed to scan db row", err))
			return
		}
		rowsDbMap[row.Timestamp] = row
	}
	// notion to db
	syncNotionToDb(rowsNotion, rowsDbMap)

	// db to notion
	syncDbToNotion(header.GetID().String(), tableId, rowsNotion)

}

func syncNotionToDb(rowsNotion []notionapi.TableRowBlock, rowsDbMap map[string]pooStatusDbEntry) {
	counter := 0
	for _, rowNotion := range rowsNotion {
		rowNotionTimestamp := rowNotion.TableRow.Cells[0][0].PlainText + "T" + rowNotion.TableRow.Cells[1][0].PlainText
		rowNotionTime, err := time.ParseInLocation("2006-01-02T15:04", rowNotionTimestamp, time.Now().Location())
		if err != nil {
			slog.Warn(fmt.Sprintln("PooRecorderSyncDb Failed to parse timestamp", err))
			return
		}
		rowNotionTimeInDbFormat := rowNotionTime.UTC().Format("2006-01-02T15:04Z07:00")
		_, exists := rowsDbMap[rowNotionTimeInDbFormat]
		if !exists {
			locationNotion := rowNotion.TableRow.Cells[3][0].PlainText
			latitude, err := strconv.ParseFloat(strings.Split(locationNotion, ",")[0], 64)
			if err != nil {
				slog.Error(fmt.Sprintln("PooRecorderSyncDb Failed to parse latitude to float", err))
				return
			}
			longitude, err := strconv.ParseFloat(strings.Split(locationNotion, ",")[1], 64)
			if err != nil {
				slog.Error(fmt.Sprintln("PooRecorderSyncDb Failed to parse longitude to float", err))
				return
			}
			_, err = db.Exec(`INSERT INTO poo_records (timestamp, status, latitude, longitude) VALUES (?, ?, ?, ?)`,
				rowNotionTimeInDbFormat, rowNotion.TableRow.Cells[2][0].PlainText, latitude, longitude)
			if err != nil {
				slog.Warn(fmt.Sprintln("PooRecorderSyncDb Failed to insert new row", err))
				return
			}
			counter++
		}
	}
	slog.Info(fmt.Sprintln("PooRecorderSyncDb Inserted", counter, "new rows from Notion to DB"))
}

func syncDbToNotion(headerId string, tableId string, rowsNotion []notionapi.TableRowBlock) {
	counter := 0
	var rowsDbSlice []pooStatusDbEntry
	rowsDb, err := db.Query(`SELECT * FROM poo_records ORDER BY timestamp DESC`)
	if err != nil {
		slog.Error(fmt.Sprintln("PooRecorderSyncDb Failed to get db rows", err))
		return
	}
	defer rowsDb.Close()
	for rowsDb.Next() {
		var row pooStatusDbEntry
		err = rowsDb.Scan(&row.Timestamp, &row.Status, &row.Latitude, &row.Longitude)
		if err != nil {
			slog.Error(fmt.Sprintln("PooRecorderSyncDb Failed to scan db row", err))
			return
		}
		rowsDbSlice = append(rowsDbSlice, row)
	}
	startFromId := headerId
	for iNotion, iDb := 0, 0; iNotion < len(rowsNotion) && iDb < len(rowsDbSlice); {
		notionTimeStamp := rowsNotion[iNotion].TableRow.Cells[0][0].PlainText + "T" + rowsNotion[iNotion].TableRow.Cells[1][0].PlainText
		notionTime, err := time.ParseInLocation("2006-01-02T15:04", notionTimeStamp, time.Now().Location())
		if err != nil {
			slog.Warn(fmt.Sprintln("PooRecorderSyncDb Failed to parse notion timestamp", err))
			return
		}
		notionTimeStampInDbFormat := notionTime.UTC().Format("2006-01-02T15:04Z07:00")
		dbTimeStamp := rowsDbSlice[iDb].Timestamp
		dbTime, err := time.Parse("2006-01-02T15:04Z07:00", dbTimeStamp)
		if err != nil {
			slog.Warn(fmt.Sprintln("PooRecorderSyncDb Failed to parse db timestamp", err))
			return
		}
		dbTimeLocal := dbTime.Local()
		dbTimeDate := dbTimeLocal.Format("2006-01-02")
		dbTimeTime := dbTimeLocal.Format("15:04")
		if notionTimeStampInDbFormat == dbTimeStamp {
			startFromId = rowsNotion[iNotion].GetID().String()
			iNotion++
			iDb++
			continue
		}
		if iNotion != len(rowsNotion)-1 {
			notionNextTimeStamp := rowsNotion[iNotion+1].TableRow.Cells[0][0].PlainText + "T" + rowsNotion[iNotion+1].TableRow.Cells[1][0].PlainText
			notionNextTime, err := time.ParseInLocation("2006-01-02T15:04", notionNextTimeStamp, time.Now().Location())
			if err != nil {
				slog.Warn(fmt.Sprintln("PooRecorderSyncDb Failed to parse next notion timestamp", err))
				return
			}
			if notionNextTime.After(notionTime) {
				slog.Error(fmt.Sprintf("PooRecorderSyncDb Notion timestamp %s is after next timestamp %s, checking, aborting", notionTimeStamp, notionNextTimeStamp))
				return
			}
		}
		id, err := notion.WriteTableRow([]string{
			dbTimeDate,
			dbTimeTime,
			rowsDbSlice[iDb].Status,
			fmt.Sprintf("%f,%f",
				rowsDbSlice[iDb].Latitude, rowsDbSlice[iDb].Longitude)},
			tableId,
			startFromId)
		if err != nil {
			slog.Warn(fmt.Sprintln("PooRecorderSyncDb Failed to write row to Notion", err))
			return
		}
		startFromId = id
		iDb++
		counter++
		time.Sleep(400 * time.Millisecond)
	}
	slog.Info(fmt.Sprintln("PooRecorderSyncDb Inserted", counter, "new rows from DB to Notion"))
}

func storeStatus(record recordDetail, timestamp time.Time) error {
	tableId := viper.GetString("pooRecorder.tableId")
	recordDate := timestamp.Format("2006-01-02")
	recordTime := timestamp.Format("15:04")
	slog.Debug(fmt.Sprintln("Recording poo", record.Status, "at", record.Latitude, record.Longitude))
	_, err := db.Exec(`INSERT OR IGNORE INTO poo_records (timestamp, status, latitude, longitude) VALUES (?, ?, ?, ?)`,
		timestamp.UTC().Format("2006-01-02T15:04Z07:00"), record.Status, record.Latitude, record.Longitude)
	if err != nil {
		return err
	}
	go func() {
		header, err := notion.GetTableRows(tableId, 1, "")
		if err != nil {
			slog.Warn(fmt.Sprintln("HandleRecordPoo Failed to get table header", err))
			return
		}
		if len(header) == 0 {
			slog.Warn("HandleRecordPoo Table header not found")
			return
		}
		headerId := header[0].GetID()
		_, err = notion.WriteTableRow([]string{recordDate, recordTime, record.Status, record.Latitude + "," + record.Longitude}, tableId, headerId.String())
		if err != nil {
			slog.Warn(fmt.Sprintln("HandleRecordPoo Failed to write table row", err))
		}
	}()
	return nil
}