feat(projections): resource counters (#9979)

# Which Problems Are Solved

Add the ability to keep track of the current counts of projection
resources. We want to prevent calling `SELECT COUNT(*)` on tables, as
that forces a full scan and sudden spikes of DB resource uses.

# How the Problems Are Solved

- A resource_counts table is added
- Triggers that increment and decrement the counted values on inserts
and deletes
- Triggers that delete all counts of a table when the source table is
TRUNCATEd. This is not in the business logic, but prevents wrong counts
in case someone want to force a re-projection.
- Triggers that delete all counts if the parent resource is deleted
- Script to pre-populate the resource_counts table when a new source
table is added.

The triggers are reusable for any type of resource, in case we choose to
add more in the future.
Counts are aggregated by a given parent. Currently only `instance` and
`organization` are defined as possible parent. This can later be
extended to other types, such as `project`, should the need arise.

I deliberately chose to use `parent_id` to distinguish from the
de-factor `resource_owner` which is usually an organization ID. For
example:

- For users the parent is an organization and the `parent_id` matches
`resource_owner`.
- For organizations the parent is an instance, but the `resource_owner`
is the `org_id`. In this case the `parent_id` is the `instance_id`.
- Applications would have a similar problem, where the parent is a
project, but the `resource_owner` is the `org_id`


# Additional Context

Closes https://github.com/zitadel/zitadel/issues/9957

internal/migration/trigger.go

@@ -0,0 +1,127 @@
+package migration
+
+import (
+	"context"
+	"embed"
+	"fmt"
+	"strings"
+	"text/template"
+
+	"github.com/mitchellh/mapstructure"
+
+	"github.com/zitadel/zitadel/internal/database"
+	"github.com/zitadel/zitadel/internal/domain"
+	"github.com/zitadel/zitadel/internal/eventstore"
+)
+
+const (
+	countTriggerTmpl       = "count_trigger"
+	deleteParentCountsTmpl = "delete_parent_counts_trigger"
+)
+
+var (
+	//go:embed *.sql
+	templateFS embed.FS
+	templates  = template.Must(template.ParseFS(templateFS, "*.sql"))
+)
+
+// CountTrigger registers the existing projections.count_trigger function.
+// The trigger than takes care of keeping count of existing
+// rows in the source table.
+// It also pre-populates the projections.resource_counts table with
+// the counts for the given table.
+//
+// During the population of the resource_counts table,
+// the source table is share-locked to prevent concurrent modifications.
+// Projection handlers will be halted until the lock is released.
+// SELECT statements are not blocked by the lock.
+//
+// This migration repeats when any of the arguments are changed,
+// such as renaming of a projection table.
+func CountTrigger(
+	db *database.DB,
+	table string,
+	parentType domain.CountParentType,
+	instanceIDColumn string,
+	parentIDColumn string,
+	resource string,
+) RepeatableMigration {
+	return &triggerMigration{
+		triggerConfig: triggerConfig{
+			Table:            table,
+			ParentType:       parentType.String(),
+			InstanceIDColumn: instanceIDColumn,
+			ParentIDColumn:   parentIDColumn,
+			Resource:         resource,
+		},
+		db:           db,
+		templateName: countTriggerTmpl,
+	}
+}
+
+// DeleteParentCountsTrigger
+//
+// This migration repeats when any of the arguments are changed,
+// such as renaming of a projection table.
+func DeleteParentCountsTrigger(
+	db *database.DB,
+	table string,
+	parentType domain.CountParentType,
+	instanceIDColumn string,
+	parentIDColumn string,
+	resource string,
+) RepeatableMigration {
+	return &triggerMigration{
+		triggerConfig: triggerConfig{
+			Table:            table,
+			ParentType:       parentType.String(),
+			InstanceIDColumn: instanceIDColumn,
+			ParentIDColumn:   parentIDColumn,
+			Resource:         resource,
+		},
+		db:           db,
+		templateName: deleteParentCountsTmpl,
+	}
+}
+
+type triggerMigration struct {
+	triggerConfig
+	db           *database.DB
+	templateName string
+}
+
+// String implements [Migration] and [fmt.Stringer].
+func (m *triggerMigration) String() string {
+	return fmt.Sprintf("repeatable_%s_%s", m.Resource, m.templateName)
+}
+
+// Execute implements [Migration]
+func (m *triggerMigration) Execute(ctx context.Context, _ eventstore.Event) error {
+	var query strings.Builder
+	err := templates.ExecuteTemplate(&query, m.templateName, m.triggerConfig)
+	if err != nil {
+		return fmt.Errorf("%s: execute trigger template: %w", m, err)
+	}
+	_, err = m.db.ExecContext(ctx, query.String())
+	if err != nil {
+		return fmt.Errorf("%s: exec trigger query: %w", m, err)
+	}
+	return nil
+}
+
+type triggerConfig struct {
+	Table            string `json:"table,omitempty" mapstructure:"table"`
+	ParentType       string `json:"parent_type,omitempty" mapstructure:"parent_type"`
+	InstanceIDColumn string `json:"instance_id_column,omitempty" mapstructure:"instance_id_column"`
+	ParentIDColumn   string `json:"parent_id_column,omitempty" mapstructure:"parent_id_column"`
+	Resource         string `json:"resource,omitempty" mapstructure:"resource"`
+}
+
+// Check implements [RepeatableMigration].
+func (c *triggerConfig) Check(lastRun map[string]any) bool {
+	var dst triggerConfig
+	if err := mapstructure.Decode(lastRun, &dst); err != nil {
+		panic(err)
+	}
+	return dst != *c
+}