Add identity-scoped OAuth token management

Let users create, list, and revoke manual bearer tokens from
/me/oauth-tokens, scoped to their identity rather than an
organization. Manual tokens store a null client_id and are
authorized with a self-manage IAM policy.

Wire Connect GraphQL on Identity (list, create, revoke), add
console UI with scoped create flow and credentials dialog, and
cover the flow in e2e tests. Fix list pagination ordering and
keep the Relay connection in sync after create.

Signed-off-by: Ludovic Vielle <ludovic@probo.com>

pkg/coredata/oauth2_access_token.go

@@ -18,25 +18,40 @@ import (
 	"context"
 	"errors"
 	"fmt"
+	"maps"
 	"time"
 
 	"github.com/jackc/pgx/v5"
 	"go.gearno.de/kit/pg"
 	"go.probo.inc/probo/pkg/gid"
+	"go.probo.inc/probo/pkg/iam/policy"
+	"go.probo.inc/probo/pkg/page"
 )
 
 type (
 	OAuth2AccessToken struct {
 		ID          gid.GID      `db:"id"`
+		Name        string       `db:"name"`
 		HashedValue []byte       `db:"hashed_value"`
-		ClientID    gid.GID      `db:"client_id"`
+		ClientID    *gid.GID     `db:"client_id"`
 		IdentityID  gid.GID      `db:"identity_id"`
 		Scopes      OAuth2Scopes `db:"scopes"`
 		CreatedAt   time.Time    `db:"created_at"`
 		ExpiresAt   time.Time    `db:"expires_at"`
 	}
+
+	OAuth2AccessTokens []*OAuth2AccessToken
 )
 
+func (t *OAuth2AccessToken) CursorKey(orderBy OAuth2AccessTokenOrderField) page.CursorKey {
+	switch orderBy {
+	case OAuth2AccessTokenOrderFieldCreatedAt:
+		return page.NewCursorKey(t.ID, t.CreatedAt)
+	}
+
+	panic(fmt.Sprintf("unsupported order by: %s", orderBy))
+}
+
 func (t *OAuth2AccessToken) ExpiresIn(now time.Time) time.Duration {
 	return t.ExpiresAt.Sub(now)
 }
@@ -45,6 +60,7 @@ func (t *OAuth2AccessToken) Insert(ctx context.Context, conn pg.Tx) error {
 	q := `
 INSERT INTO iam_oauth2_access_tokens (
 	id,
+	name,
 	hashed_value,
 	client_id,
 	identity_id,
@@ -53,6 +69,7 @@ INSERT INTO iam_oauth2_access_tokens (
 	expires_at
 ) VALUES (
 	@id,
+	@name,
 	@hashed_value,
 	@client_id,
 	@identity_id,
@@ -64,6 +81,7 @@ INSERT INTO iam_oauth2_access_tokens (
 
 	args := pgx.StrictNamedArgs{
 		"id":           t.ID,
+		"name":         t.Name,
 		"hashed_value": t.HashedValue,
 		"client_id":    t.ClientID,
 		"identity_id":  t.IdentityID,
@@ -80,10 +98,48 @@ INSERT INTO iam_oauth2_access_tokens (
 	return nil
 }
 
+func (t *OAuth2AccessToken) LoadByID(ctx context.Context, conn pg.Querier, id gid.GID) error {
+	q := `
+SELECT
+	id,
+	name,
+	hashed_value,
+	client_id,
+	identity_id,
+	scopes,
+	created_at,
+	expires_at
+FROM
+	iam_oauth2_access_tokens
+WHERE
+	id = @id
+LIMIT 1;
+`
+
+	rows, err := conn.Query(ctx, q, pgx.StrictNamedArgs{"id": id})
+	if err != nil {
+		return fmt.Errorf("cannot query oauth2_access_token: %w", err)
+	}
+
+	token, err := pgx.CollectExactlyOneRow(rows, pgx.RowToStructByName[OAuth2AccessToken])
+	if err != nil {
+		if errors.Is(err, pgx.ErrNoRows) {
+			return ErrResourceNotFound
+		}
+
+		return fmt.Errorf("cannot collect oauth2_access_token: %w", err)
+	}
+
+	*t = token
+
+	return nil
+}
+
 func (t *OAuth2AccessToken) LoadByHashedValue(ctx context.Context, conn pg.Querier, hashedValue []byte) error {
 	q := `
 SELECT
 	id,
+	name,
 	hashed_value,
 	client_id,
 	identity_id,
@@ -125,6 +181,7 @@ func (t *OAuth2AccessToken) LoadByHashedValueAndClientID(
 	q := `
 SELECT
 	id,
+	name,
 	hashed_value,
 	client_id,
 	identity_id,
@@ -163,6 +220,127 @@ LIMIT 1;
 	return nil
 }
 
+func (t *OAuth2AccessToken) AuthorizationAttributes(
+	ctx context.Context,
+	conn pg.Querier,
+	resourceIDs []gid.GID,
+) (policy.AttributesByID, error) {
+	q := `
+SELECT
+	t.id,
+	t.identity_id
+FROM
+	iam_oauth2_access_tokens t
+WHERE
+	t.id = ANY(@resource_ids::text[])
+`
+
+	args := pgx.StrictNamedArgs{
+		"resource_ids": resourceIDs,
+	}
+
+	rows, err := conn.Query(ctx, q, args)
+	if err != nil {
+		return nil, fmt.Errorf("cannot query oauth2 access token authorization attributes: %w", err)
+	}
+	defer rows.Close()
+
+	attrsByID := make(policy.AttributesByID, len(resourceIDs))
+
+	for rows.Next() {
+		var (
+			id         gid.GID
+			identityID gid.GID
+		)
+
+		err = rows.Scan(&id, &identityID)
+		if err != nil {
+			return nil, fmt.Errorf("cannot scan oauth2 access token authorization attributes: %w", err)
+		}
+
+		attrsByID[id] = policy.Attributes{"identity_id": identityID.String()}
+	}
+
+	if err = rows.Err(); err != nil {
+		return nil, fmt.Errorf("cannot iterate oauth2 access token authorization attributes: %w", err)
+	}
+
+	return attrsByID, nil
+}
+
+func (ts *OAuth2AccessTokens) LoadByIdentityID(
+	ctx context.Context,
+	conn pg.Querier,
+	identityID gid.GID,
+	cursor *page.Cursor[OAuth2AccessTokenOrderField],
+) error {
+	q := `
+SELECT
+	id,
+	name,
+	hashed_value,
+	client_id,
+	identity_id,
+	scopes,
+	created_at,
+	expires_at
+FROM
+	iam_oauth2_access_tokens
+WHERE
+	identity_id = @identity_id
+	AND client_id IS NULL
+	AND %s
+`
+
+	q = fmt.Sprintf(q, cursor.SQLFragment())
+
+	args := pgx.StrictNamedArgs{
+		"identity_id": identityID,
+	}
+	maps.Copy(args, cursor.SQLArguments())
+
+	rows, err := conn.Query(ctx, q, args)
+	if err != nil {
+		return fmt.Errorf("cannot query oauth2 access tokens: %w", err)
+	}
+
+	tokens, err := pgx.CollectRows(rows, pgx.RowToAddrOfStructByName[OAuth2AccessToken])
+	if err != nil {
+		return fmt.Errorf("cannot collect oauth2 access tokens: %w", err)
+	}
+
+	*ts = tokens
+
+	return nil
+}
+
+func (ts *OAuth2AccessTokens) CountByIdentityID(
+	ctx context.Context,
+	conn pg.Querier,
+	identityID gid.GID,
+) (int, error) {
+	q := `
+SELECT
+	COUNT(id)
+FROM
+	iam_oauth2_access_tokens
+WHERE
+	identity_id = @identity_id
+	AND client_id IS NULL;
+`
+
+	args := pgx.StrictNamedArgs{
+		"identity_id": identityID,
+	}
+
+	var count int
+	if err := conn.QueryRow(ctx, q, args).Scan(&count); err != nil {
+		return 0, fmt.Errorf("cannot count oauth2 access tokens: %w", err)
+	}
+
+	return count, nil
+}
+
 func (t *OAuth2AccessToken) Delete(ctx context.Context, conn pg.Tx) error {
 	q := `
 DELETE FROM iam_oauth2_access_tokens