createOplogTrigger method

@override List<String> createOplogTrigger(SqlOpType opType, QualifiedTablename table, String newPKs, String newRows, String oldRows,)

Create a trigger that logs operations into the oplog.

Implementation

@override
List<String> createOplogTrigger(
  SqlOpType opType,
  QualifiedTablename table,
  String newPKs,
  String newRows,
  String oldRows,
) {
  final namespace = table.namespace;
  final tableName = table.tablename;

  final opTypeLower = opType.text.toLowerCase();
  final pk = createPKJsonObject(newPKs);
  // Update has both the old and the new row
  // Delete only has the old row
  final newRecord =
      opType == SqlOpType.delete ? 'NULL' : createJsonbObject(newRows);
  // Insert only has the new row
  final oldRecord =
      opType == SqlOpType.insert ? 'NULL' : createJsonbObject(oldRows);

  return [
    '''
      CREATE OR REPLACE FUNCTION ${opTypeLower}_${namespace}_${tableName}_into_oplog_function()
      RETURNS TRIGGER AS \$\$
      BEGIN
        DECLARE
          flag_value INTEGER;
        BEGIN
          -- Get the flag value from _electric_trigger_settings
          SELECT flag INTO flag_value FROM "$namespace"._electric_trigger_settings WHERE namespace = '$namespace' AND tablename = '$tableName';

          IF flag_value = 1 THEN
            -- Insert into _electric_oplog
            INSERT INTO "$namespace"._electric_oplog (namespace, tablename, optype, "primaryKey", "newRow", "oldRow", timestamp)
            VALUES (
              '$namespace',
              '$tableName',
              '${opType.text}',
              $pk,
              $newRecord,
              $oldRecord,
              NULL
            );
          END IF;

          RETURN NEW;
        END;
      END;
      \$\$ LANGUAGE plpgsql;
    ''',
    '''
      CREATE TRIGGER ${opTypeLower}_${namespace}_${tableName}_into_oplog
        AFTER ${opType.text} ON $table
          FOR EACH ROW
            EXECUTE FUNCTION ${opTypeLower}_${namespace}_${tableName}_into_oplog_function();
    ''',
  ];
}