fix: Author the column default an upsert writes for a null foreign key (#164)

Co-authored-by: Cursor Agent <cursoragent@cursor.com>

packages/serverpod_offline_sync/lib/src/database/database.dart

@@ -491,9 +491,15 @@ class OfflineSyncDatabase implements Database {
           tx,
           projectionUnchanged: projection.projectionUnchanged,
           domainBeforeUpsert: projection.domain,
-          upsertRows: updateColumns == null && projection.domain.isNotEmpty
-              ? prepared.rows
-              : const [],
+          upsertInputs: updateColumns == null && projection.domain.isNotEmpty
+              ? _recorder.upsertInputsByRow(
+                  suppliedRows: prepared.rows,
+                  databaseRows: values,
+                  updatedRows: updatedRows,
+                  conflictColumns: conflictColumns,
+                )
+              : const {},
+          upsertConflictColumns: conflictColumns.map((c) => c.columnName).toSet(),
         );
         if (noReturn) return <T>[];
         _stripStampedRows(result, prepared);

packages/serverpod_offline_sync/lib/src/database/recorder.dart

@@ -782,6 +782,67 @@ class CrdtMutationRecorder {
       }),
   ];
 
+  /// Associates caller values with the rows accepted by the physical upsert.
+  ///
+  /// A unique conflict target can select an existing row with a different ID.
+  /// Match its unchanged conflict columns against the default-resolved inputs;
+  /// returned positions cannot be used because `updateWhere` can skip inputs.
+  Map<MergeRowKey, Map<String, Object?>> upsertInputsByRow<T extends TableRow>({
+    required List<T> suppliedRows,
+    required List<T> databaseRows,
+    required List<T> updatedRows,
+    required List<Column> conflictColumns,
+  }) {
+    if (updatedRows.isEmpty) return const {};
+
+    if (conflictColumns.length == 1 && conflictColumns.single.columnName == 'id') {
+      final inputs = <MergeRowKey, Map<String, Object?>>{};
+      for (final row in suppliedRows) {
+        if (row.id case final UuidValue rowId) {
+          inputs.putIfAbsent(
+            (row.table.tableName, rowId),
+            () => _authoredValuesFromRow(row, null),
+          );
+        }
+      }
+      return inputs;
+    }
+
+    final inputsByConflictKey = <String, Map<String, Object?>>{};
+    for (final (index, row) in databaseRows.indexed) {
+      final key = _upsertConflictKey(row, conflictColumns);
+      if (key == null) continue;
+
+      // A repeated target may appear after an accepted update makes
+      // `updateWhere` false. Keep the first input, not the later skipped one;
+      // the delegate rejects batches that accept the same target twice.
+      inputsByConflictKey.putIfAbsent(
+        key,
+        () => _authoredValuesFromRow(suppliedRows[index], null),
+      );
+    }
+    return {
+      for (final row in updatedRows)
+        (row.table.tableName, row.id as UuidValue):
+            ?inputsByConflictKey[_upsertConflictKey(row, conflictColumns)],
+    };
+  }
+
+  String? _upsertConflictKey(TableRow row, List<Column> conflictColumns) {
+    final values = row.toJsonForDatabase() as Map;
+    final parts = <String>[];
+    for (final column in conflictColumns) {
+      final value = values[column.columnName];
+      // A null component makes no unique claim and cannot select an update.
+      if (value == null) return null;
+
+      final part = canonicalProjectionValue(value);
+      // Length prefixes preserve component boundaries even inside text keys.
+      parts.add('${part.length}:$part');
+    }
+    return parts.join();
+  }
+
   Future<Map<UuidValue, Map<String, Object?>>> _readPlannedDomainValues(
     String tableName,
     Set<UuidValue> rowIds,
@@ -944,7 +1005,8 @@ class CrdtMutationRecorder {
     bool authoredColumnValues = false,
     Set<MergeRowKey> projectionRows = const {},
     Map<MergeRowKey, Map<String, Object?>> domainBeforeUpsert = const {},
-    List<TableRow> upsertRows = const [],
+    Map<MergeRowKey, Map<String, Object?>> upsertInputs = const {},
+    Set<String> upsertConflictColumns = const {},
   }) async {
     await _foreignKeyProjector.assertVisibleTargets(updatedRows, columns, transaction);
 
@@ -953,6 +1015,16 @@ class CrdtMutationRecorder {
       rowIds,
       _,
     ) async {
+      // Serverpod's full-row upsert updates only non-conflict columns. Keep
+      // the caller's null `columns` separate from this physical write set so
+      // unchanged projected values still follow full-row passthrough rules.
+      final writtenColumns =
+          columns ??
+          (upsertConflictColumns.isEmpty
+              ? null
+              : updatedRows.first.table.managedColumns
+                    .where((c) => !upsertConflictColumns.contains(c.columnName))
+                    .toList());
       final crdtDataRows = await _context.findRequiredCrdtRows(
         tableName,
         rowIds,
@@ -966,24 +1038,13 @@ class CrdtMutationRecorder {
               transaction: transaction,
             )
           : const <MergeFieldKey>{};
-      final upsertInputs = upsertRows.isEmpty || implicitForeignKeyRepairFields.isEmpty
-          ? const <MergeRowKey, Map<String, Object?>>{}
-          : {
-              for (final row in upsertRows)
-                if (row.id case final UuidValue rowId)
-                  (row.table.tableName, rowId):
-                      row.toJsonForDatabase() as Map<String, dynamic>,
-            };
-      // Insert defaults must not author over an echoed null that is holding
-      // a projected FK claim.
-      bool echoesProjectedNull(UuidValue rowId, String columnName) {
-        if (!implicitForeignKeyRepairFields.contains((tableName, rowId, columnName))) {
-          return false;
-        }
+      // A null foreign key is resolved to its column default before the
+      // physical upsert, so the stored value can equal a projected default the
+      // caller never wrote. Passthrough is about what the caller supplied.
+      Object? suppliedValue(UuidValue rowId, String columnName, Object? stored) {
         final input = upsertInputs[(tableName, rowId)];
-        return input != null &&
-            input[columnName] == null &&
-            domainBeforeUpsert[(tableName, rowId)]?[columnName] == null;
+        if (input == null || !input.containsKey(columnName)) return stored;
+        return input[columnName];
       }
 
       // Explicit columns author even an unchanged null; full-row passthrough
@@ -992,7 +1053,7 @@ class CrdtMutationRecorder {
         if (authoredColumnValues)
           for (final row in updatedRows)
             for (final MapEntry(key: columnName, value: value)
-                in _authoredValuesFromRow(row, columns).entries)
+                in _authoredValuesFromRow(row, writtenColumns).entries)
               (tableName, row.id as UuidValue, columnName): value
         else
           // Only rows returned by the physical upsert were accepted by its
@@ -1000,22 +1061,21 @@ class CrdtMutationRecorder {
           for (final row in updatedRows)
             if (domainBeforeUpsert.containsKey((tableName, row.id)))
               for (final MapEntry(key: columnName, value: value)
-                  in _authoredValuesFromRow(row, columns).entries)
+                  in _authoredValuesFromRow(row, writtenColumns).entries)
                 if ((domainBeforeUpsert[(tableName, row.id)]?.containsKey(columnName) ??
                         false) &&
                     (columns != null ||
-                        (!projectionValuesEqual(
-                              domainBeforeUpsert[(tableName, row.id)]![columnName],
-                              value,
-                            ) &&
-                            !echoesProjectedNull(row.id as UuidValue, columnName))))
+                        !projectionValuesEqual(
+                          domainBeforeUpsert[(tableName, row.id)]![columnName],
+                          suppliedValue(row.id as UuidValue, columnName, value),
+                        )))
                   (tableName, row.id as UuidValue, columnName): value,
       };
       _uniqueResolver.validateAuthoredFields(authoredOverlays);
       await _recordUpdatedFields(
         updatedRows,
         crdtDataRows,
-        columns,
+        writtenColumns,
         transaction,
         skippedFields: implicitForeignKeyRepairFields.difference(
           authoredOverlays.keys.toSet(),
@@ -1026,11 +1086,13 @@ class CrdtMutationRecorder {
             updatedRows,
             transaction,
             inserting: false,
-            columns: columns,
+            columns: writtenColumns,
           )) {
         return;
       }
-      final updatedColumnNames = columns?.map((column) => column.columnName).toSet();
+      final updatedColumnNames = writtenColumns
+          ?.map((column) => column.columnName)
+          .toSet();
       await _maybeProject(
         tableName,
         rowIds,

packages/serverpod_offline_sync_server/test/integration/untracked_update_test.dart

@@ -24,7 +24,22 @@ void main() {
   });
 
   withServerpod(
-    'PostgreSQL untracked updates with the database interceptor',
+    '[PostgreSQL untracked updates with the database interceptor]',
+    databaseInterceptor: offlineSyncDatabaseInterceptor,
+    serverDirectory: serverDirectory,
+    configOverride: (config) => config.copyWith(
+      apiServer: ServerConfig(
+        port: 0,
+        publicHost: 'localhost',
+        publicPort: 0,
+        publicScheme: 'http',
+      ),
+      database: PostgresDatabaseConfig.embedded(
+        dataPath: '${Directory.systemTemp.path}/offline_sync_updates_postgres_$pid',
+        name: 'serverpod_test',
+        maxConnectionCount: 5,
+      ),
+    ),
     (sessionBuilder, _) {
       late Session session;
 
@@ -65,21 +80,6 @@ void main() {
         });
       });
     },
-    databaseInterceptor: offlineSyncDatabaseInterceptor,
-    serverDirectory: serverDirectory,
-    configOverride: (config) => config.copyWith(
-      apiServer: ServerConfig(
-        port: 0,
-        publicHost: 'localhost',
-        publicPort: 0,
-        publicScheme: 'http',
-      ),
-      database: PostgresDatabaseConfig.embedded(
-        dataPath: '${Directory.systemTemp.path}/offline_sync_updates_postgres_$pid',
-        name: 'serverpod_test',
-        maxConnectionCount: 5,
-      ),
-    ),
   );
 }