Method GetOrCreateSource

internal/store/sync.go:578–620

GetOrCreateSource gets or creates a source by type and identifier. Concurrent first-inserts converge via INSERT ... ON CONFLICT DO UPDATE RETURNING: the no-op SET fires RETURNING on both the insert and conflict path so the second caller receives the existing row's fields instead of a unique-violatio

(sourceType, identifier string)

// conflict path so the second caller receives the existing row's
// fields instead of a unique-violation error.
func (s *Store) GetOrCreateSource(sourceType, identifier string) (*Source, error) {
	now := s.dialect.Now()
	row := s.db.QueryRow(fmt.Sprintf(`
		INSERT INTO sources (source_type, identifier, created_at, updated_at)
		VALUES (?, ?, %s, %s)
		ON CONFLICT (source_type, identifier) DO UPDATE
		SET identifier = sources.identifier
		RETURNING id, source_type, identifier, display_name, google_user_id,
		          last_sync_at, sync_cursor, sync_config, oauth_app,
		          created_at, updated_at
	`, now, now), sourceType, identifier)

	source, err := scanSource(row)
	if err != nil {
		return nil, fmt.Errorf("upsert source: %w", err)
	}

	// Add to the default "All" collection if it exists.
	//
	// This runs as a separate Exec rather than inside a transaction
	// with the source insert. If this Exec fails, the source row is
	// committed but the All membership is missing — and the next
	// EnsureDefaultCollection call (which runs in InitSchema on every
	// process launch) re-adds every source not yet linked. Self-heals
	// on next CLI invocation; until then collection-scoped reads of
	// All would miss this source. Acceptable for a single-user tool;
	// a future refactor can fold this into a withTx.
	if _, err := s.db.Exec(
		s.dialect.InsertOrIgnore(
			`INSERT OR IGNORE INTO collection_sources (collection_id, source_id)
			 SELECT id, ? FROM collections WHERE name = ?`,
		),
		source.ID, DefaultCollectionName,
	); err != nil {
		slog.Warn("failed to add source to default collection (self-heals on next InitSchema)",
			"source_id", source.ID,
			"identifier", identifier,
			"error", err,
		)
	}

	return source, nil
}

// UpdateSourceSyncCursor updates the sync cursor (historyId) for a source.
func (s *Store) UpdateSourceSyncCursor(sourceID int64, cursor string) error {