Function executeWriteTuples

pkg/storage/postgres/postgres.go:491–538

For the prepared writeItems, execute insert writeItems.

(ctx context.Context, txn PgxExec, writeItems [][]interface{})

// For the prepared writeItems, execute insert writeItems.
func executeWriteTuples(ctx context.Context, txn PgxExec, writeItems [][]interface{}) error {
	stbl := sq.StatementBuilder.PlaceholderFormat(sq.Dollar)

	for start, totalWrites := 0, len(writeItems); start < totalWrites; start += storage.DefaultMaxTuplesPerWrite {
		end := start + storage.DefaultMaxTuplesPerWrite
		if end > totalWrites {
			end = totalWrites
		}

		writesBatch := writeItems[start:end]

		insertBuilder := stbl.
			Insert("tuple").
			Columns(
				"store",
				"object_type",
				"object_id",
				"relation",
				"_user",
				"user_type",
				"condition_name",
				"condition_context",
				"ulid",
				"inserted_at",
			)

		for _, item := range writesBatch {
			insertBuilder = insertBuilder.Values(item...)
		}

		stmt, args, err := insertBuilder.ToSql()
		if err != nil {
			// Should never happen because we craft the insert statement
			return HandleSQLError(err)
		}

		_, err = txn.Exec(ctx, stmt, args...)
		if err != nil {
			dberr := HandleSQLError(err)
			if errors.Is(dberr, storage.ErrCollision) {
				// ErrCollision is returned on duplicate write (constraint violation), meaning we hit a race condition - someone else inserted the same row(s).
				return storage.ErrWriteConflictOnInsert
			}
			return dberr
		}
	}
	return nil
}

func executeInsertChanges(ctx context.Context, txn PgxExec, changeLogItems [][]interface{}) error {
	stbl := sq.StatementBuilder.PlaceholderFormat(sq.Dollar)