Adding log table cleanup based on snapshotting

This will allow configuring snashopt cleanup via log compaction.

db/changelog.go

@@ -53,6 +53,30 @@ func init() {
     )
 }
 
+func (conn *SqliteStreamDB) Replicate(event *ChangeLogEvent) error {
+    if err := conn.consumeReplicationEvent(event); err != nil {
+        return err
+    }
+    return nil
+}
+
+func (conn *SqliteStreamDB) CleanupChangeLogs() error {
+    for name := range conn.watchTablesSchema {
+        log.Debug().Str("table", name).Msg("Cleaning up change logs")
+        metaTableName := conn.metaTable(name, changeLogName)
+        _, err := conn.Delete(metaTableName).
+            Where(goqu.Ex{"state": Published}).
+            Prepared(true).
+            Executor().
+            Exec()
+
+        if err != nil {
+            return err
+        }
+    }
+    return nil
+}
+
 func (conn *SqliteStreamDB) tableCDCScriptFor(tableName string) (string, error) {
     columns, ok := conn.watchTablesSchema[tableName]
     if !ok {
@@ -74,13 +98,6 @@ func (conn *SqliteStreamDB) tableCDCScriptFor(tableName string) (string, error)
     return spaceStripper.ReplaceAllString(buf.String(), "\n    "), nil
 }
 
-func (conn *SqliteStreamDB) Replicate(event *ChangeLogEvent) error {
-    if err := conn.consumeReplicationEvent(event); err != nil {
-        return err
-    }
-    return nil
-}
-
 func (conn *SqliteStreamDB) consumeReplicationEvent(event *ChangeLogEvent) error {
     return conn.WithTx(func(tnx *goqu.TxDatabase) error {
         primaryKeyMap := conn.getPrimaryKeyMap(event)

lib/replica_machine.go

@@ -59,6 +59,11 @@ func (ssm *SQLiteStateMachine) Lookup(_ interface{}) (interface{}, error) {
 }
 
 func (ssm *SQLiteStateMachine) SaveSnapshot(_ io.Writer, _ sm.ISnapshotFileCollection, _ <-chan struct{}) error {
+    err := ssm.DB.CleanupChangeLogs()
+    if err != nil {
+        return err
+    }
+
     return nil
 }