Phase 3 (a) add sharding key routing to LFK triggers

What does this MR do and why?

Add sharding key routing to loose foreign key triggers

Phase 3a of making loose_foreign_keys_deleted_records compatible with Cells and the Organization Mover.

The shared trigger functions now accept an optional JSONB argument and route deleted records to the matching sharding-key table (organization, namespace, project, user).

With no argument, or when all sharding key values are NULL, records fall back to the cell-local table. Existing triggers pass no argument, so behavior is unchanged, and no trigger is rewritten here.

Adds LooseForeignKeyHelpers sharding_keys_for/sharding_keys_args and the temporary track_record_deletions_with_sharding_keys helpers to wire up tables in later phases.

Phase	MR
1 - Foundation	!233234 (merged)
2 (a) - Inject record_store into LFK services	!237757 (merged)
2 (b) - DeletedRecordStore facade + workers + feature flag	!236671 (merged)
3 (a) - Trigger function update	This MR
3 (b) - First table rollout	-
4 - Gradual trigger rollout	-
5 - Cleanup	-

How to set up and validate locally

1. Let's create the setup and check if the migration helper reads the correct sharding keys from db/docs:

m = ActiveRecord::Migration.new.extend(Gitlab::Database::MigrationHelpers::LooseForeignKeyHelpers)
conn = ApplicationRecord.connection

STORES = [
  LooseForeignKeys::DeletedRecord,
  LooseForeignKeys::OrganizationDeletedRecord,
  LooseForeignKeys::NamespaceDeletedRecord,
  LooseForeignKeys::ProjectDeletedRecord,
  LooseForeignKeys::UserDeletedRecord
]

report = ->(table) do
  Gitlab::Database::SharedModel.using_connection(ApplicationRecord.connection) do
    STORES.to_h do |s|
      [s.table_name.sub('loose_foreign_keys_', ''), s.where(fully_qualified_table_name: "public.#{table}").count]
    end
  end
end

batch = ->(table) do
  Gitlab::Database::SharedModel.using_connection(ApplicationRecord.connection) do
    Gitlab::LooseForeignKeys::DeletedRecordStore.load_batch_for_table("public.#{table}", 100).map { |r| [r.class.table_name.sub('loose_foreign_keys_', ''), r.primary_key_value] }
  end
end

%w[project_repositories notes].each { |t| puts "#{t}: #{m.sharding_keys_for(t)}" }

You should get:

project_repositories: [{:table=>"loose_foreign_keys_project_deleted_records", :column=>"project_id", :source=>"project_id"}]
notes: [{:table=>"loose_foreign_keys_namespace_deleted_records", :column=>"namespace_id", :source=>"namespace_id"}, {:table=>"loose_foreign_keys_project_deleted_records", :column=>"project_id", :source=>"project_id"}, {:table=>"loose_foreign_keys_organization_deleted_records", :column=>"organization_id", :source=>"organization_id"}]

2. Keeps records cell-local when the trigger gets no sharding keys argument (current production behavior)

ActiveRecord::Base.transaction do
  ids = conn.select_values("SELECT id FROM project_repositories ORDER BY id LIMIT 2")
  conn.execute("DELETE FROM project_repositories WHERE id IN (#{ids.join(',')})")

  puts report.call('project_repositories').inspect

  LooseForeignKeys::ProcessDeletedRecordsService.new(connection: conn, logger: Logger.new(IO::NULL), record_store: Gitlab::LooseForeignKeys::DeletedRecordStore).execute

  status = Gitlab::Database::SharedModel.using_connection(ApplicationRecord.connection) do
    LooseForeignKeys::DeletedRecord.where(fully_qualified_table_name: 'public.project_repositories').group(:status).count
  end
  
  puts status.inspect

  raise ActiveRecord::Rollback
end

You should get:

{"deleted_records"=>2, "organization_deleted_records"=>0, "namespace_deleted_records"=>0, "project_deleted_records"=>0, "user_deleted_records"=>0}
{"processed"=>2}

• 2 entries in deleted_records
• 2 records consumed by the LooseForeignKeys::ProcessDeletedRecordsService

3. Routes correctly for records that has only 1 sharding key

It creates the trigger for project_repositories and the deleted records are routed to the loose_foreign_keys_project_deleted_records table

ActiveRecord::Base.transaction do
  m.track_record_deletions_with_sharding_keys(:project_repositories)

  ids = conn.select_values("SELECT id FROM project_repositories ORDER BY id LIMIT 2")
  conn.execute("DELETE FROM project_repositories WHERE id IN (#{ids.join(',')})")

  puts report.call('project_repositories').inspect
  puts batch.call('project_repositories').inspect

  raise ActiveRecord::Rollback
end

You should get:

{"deleted_records"=>0, "organization_deleted_records"=>0, "namespace_deleted_records"=>0, "project_deleted_records"=>2, "user_deleted_records"=>0}
[["project_deleted_records", 1], ["project_deleted_records", 2]]

• 2 entries in project_deleted_records
• 2 loaded records from Gitlab::LooseForeignKeys::DeletedRecordStore batch

4. Routes correctly for records that has 3 sharding keys

ActiveRecord::Base.transaction do
  m.track_record_deletions_with_sharding_keys(:notes)

  id = conn.select_value("SELECT id FROM notes WHERE project_id IS NOT NULL ORDER BY id LIMIT 1")
  conn.execute(<<~SQL)
    UPDATE notes 
    SET namespace_id = (SELECT id FROM namespaces ORDER BY id LIMIT 1),
        organization_id = (SELECT id FROM organizations ORDER BY id LIMIT 1)
    WHERE id = #{id}
  SQL
  
  conn.execute("DELETE FROM notes WHERE id = #{id}")

  puts report.call('notes').inspect
  puts batch.call('notes').inspect
  
  raise ActiveRecord::Rollback
end

You should get:

{"deleted_records"=>0, "organization_deleted_records"=>1, "namespace_deleted_records"=>1, "project_deleted_records"=>1, "user_deleted_records"=>0}
[["namespace_deleted_records", 1], ["organization_deleted_records", 1], ["project_deleted_records", 1]]

• 3 entries: one in organization_deleted_records, one in namespace_deleted_records and another in project_deleted_records
• 3 loaded records from Gitlab::LooseForeignKeys::DeletedRecordStore batch

MR acceptance checklist

Evaluate this MR against the MR acceptance checklist. It helps you analyze changes to reduce risks in quality, performance, reliability, security, and maintainability.

Related #597949