package compose import ( "cmp" "context" "fmt" "maps" "slices" "sort" "strings" apierrors "k8s.io/apimachinery/pkg/api/errors" "k8s.io/apimachinery/pkg/api/meta" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" commonv1 "github.com/johnandersen777/socialweb-computer-kcp/api/common/v1alpha1" pdsv1alpha1 "github.com/johnandersen777/socialweb-computer-kcp/api/pds/v1alpha1" "github.com/johnandersen777/socialweb-computer-kcp/api/v1alpha1" "github.com/johnandersen777/socialweb-computer-kcp/internal/contract/atprotorepo" "github.com/johnandersen777/socialweb-computer-kcp/internal/contract/market" "github.com/johnandersen777/socialweb-computer-kcp/internal/contract/vault" "github.com/johnandersen777/socialweb-computer-kcp/internal/controllers/runtime/kube" ) const SessionOwnerLabel = market.SessionOwnerLabel type SessionOwner struct { DID string Vault string Scope string } type SessionOwnerState struct { PdsURL string Ready bool Message string } func (o SessionOwner) did() string { return strings.TrimSpace(o.DID) } func (o SessionOwner) sessionPath() string { return "session/" + o.did() } func (o SessionOwner) sessionScope() string { return "auth/" + AtprotoMountPath + "/" + o.sessionPath() } func (o SessionOwner) SessionRef(r Root) commonv1.SecretRef { return commonv1.SecretRef{Vault: &commonv1.ObjectRef{Group: v1alpha1.GroupName, Kind: KindVault, Name: o.Vault, Workspace: r.Workspace}, Mount: "auth/" + AtprotoMountPath, Path: o.sessionPath()} } func ownerPdsName(pdsURL string) string { return "owner-" + Stamp(strings.TrimRight(strings.TrimSpace(pdsURL), "/")) } func SessionOwnerPds(pdsURL, ws string) (commonv1.ObjectRef, string) { return RemotePDSRefIn(ownerPdsName(pdsURL), ws), pdsServiceDID(pdsURL) } func sessionOwnerLabels() map[string]string { return map[string]string{SessionOwnerLabel: "true"} } func SessionOwnerLinks(r Root, ws, contracts string, o SessionOwner, pdsURL string) []Link { identity := AppRef(KindMarketIdentity, OwnerIdentityName(o.did()), ws) links := []Link{RefLink(identity, VaultRefIn(o.Vault, r.Workspace), Cap(vault.CapabilityAtprotoSession, o.sessionScope()))} if pdsURL = strings.TrimRight(strings.TrimSpace(pdsURL), "/"); pdsURL == "" { return links } repo := RepoRefIn(o.did(), ws) remote := AppRef(KindRemoteRepo, repo.Name, ws) links = append(links, RepoBackendLink(repo, remote)) links = append(links, RemoteRepoLinks(remote, RemotePDSRefIn(ownerPdsName(pdsURL), ws), identity, o.did())...) return append(links, InfraLinks(contracts, repo)...) } func (c *Composer) SessionOwnerState(ctx context.Context, ws, did string) (SessionOwnerState, error) { name := OwnerIdentityName(did) id, err := c.Identity(ctx, ws, name) if apierrors.IsNotFound(err) { return SessionOwnerState{Message: "the MarketIdentity " + name + " is not applied yet"}, nil } if err != nil { return SessionOwnerState{}, err } return SessionOwnerStateOf(id, did), nil } func SessionOwnerStateOf(id *v1alpha1.MarketIdentity, did string) SessionOwnerState { name := OwnerIdentityName(did) if id.Spec.Mode != v1alpha1.MarketIdentityModeOAuthSessionProxy || strings.TrimSpace(id.Status.Did) != strings.TrimSpace(did) { return SessionOwnerState{Message: "the MarketIdentity " + name + " has not reported " + did + " yet"} } st := SessionOwnerState{Message: cmp.Or(id.Status.Message, string(id.Status.Phase)), Ready: id.Status.Phase == v1alpha1.MarketIdentityPhaseReady && id.Status.ObservedGeneration >= id.Generation} switch id.Status.Phase { case v1alpha1.MarketIdentityPhaseReady, v1alpha1.MarketIdentityPhaseSessionExpired: st.PdsURL = strings.TrimRight(strings.TrimSpace(id.Status.PdsURL), "/") } return st } const ( ConditionSessionOwnersReady = "SessionOwnersReady" ReasonSessionOwnersReady = "Ready" ReasonSessionOwnerNotReady = "OwnerNotReady" ) func SetSessionOwnersCondition(conditions *[]metav1.Condition, gen int64, states map[string]SessionOwnerState) { var waiting []string for _, did := range slices.Sorted(maps.Keys(states)) { if st := states[did]; !st.Ready { waiting = append(waiting, did+": "+st.Message) } } if len(waiting) == 0 { meta.SetStatusCondition(conditions, metav1.Condition{Type: ConditionSessionOwnersReady, Status: metav1.ConditionTrue, Reason: ReasonSessionOwnersReady, Message: fmt.Sprintf("%d tenant owner(s) publish Infra through the web app session", len(states)), ObservedGeneration: gen}) return } meta.SetStatusCondition(conditions, metav1.Condition{Type: ConditionSessionOwnersReady, Status: metav1.ConditionFalse, Reason: ReasonSessionOwnerNotReady, Message: strings.Join(waiting, "; "), ObservedGeneration: gen}) } func (c *Composer) ApplySessionOwner(ctx context.Context, r Root, ws string, o SessionOwner, pdsURL string) error { ref := o.SessionRef(r) name := OwnerIdentityName(o.did()) identity := &v1alpha1.MarketIdentity{ObjectMeta: metav1.ObjectMeta{Name: name, Labels: sessionOwnerLabels()}, Spec: v1alpha1.MarketIdentitySpec{Mode: v1alpha1.MarketIdentityModeOAuthSessionProxy, Did: o.did(), Association: v1alpha1.AssociationSkip, SessionRef: &ref, SessionScope: strings.TrimSpace(o.Scope)}} if err := c.ApplyTyped(ctx, r, ws, MarketIdentityGVR, v1alpha1.GroupVersion.String(), KindMarketIdentity, identity); err != nil { return err } if pdsURL = strings.TrimRight(strings.TrimSpace(pdsURL), "/"); pdsURL == "" { return nil } remote := pdsv1alpha1.RemotePDSSpec{URL: pdsURL, DID: pdsServiceDID(pdsURL)} if _, err := c.ApplyRemotePDS(ctx, r, ws, &pdsv1alpha1.RemotePDS{ObjectMeta: metav1.ObjectMeta{Name: ownerPdsName(pdsURL), Labels: sessionOwnerLabels()}, Spec: remote}); err != nil { return err } repo := RepoRefIn(o.did(), ws).Name spec := v1alpha1.RemoteRepoSpec{DID: o.did(), PdsRef: RemotePDSRefIn(ownerPdsName(pdsURL), ""), CredentialsRef: commonv1.CredentialsRef{ObjectRef: commonv1.ObjectRef{Group: v1alpha1.GroupName, Kind: KindMarketIdentity, Name: name}}} return c.applyRepo(ctx, r, ws, repo, KindRemoteRepo, v1alpha1.RemoteRepoGVR, &v1alpha1.RemoteRepo{ObjectMeta: metav1.ObjectMeta{Name: repo, Labels: sessionOwnerLabels()}, Spec: spec}, sessionOwnerLabels()) } func (c *Composer) SessionOwners(ctx context.Context, ws string) ([]string, error) { ri, err := c.Client(ws, MarketIdentityGVR) if err != nil { return nil, err } items, err := kube.List[v1alpha1.MarketIdentity](ctx, ri, metav1.ListOptions{LabelSelector: SessionOwnerLabel + "=true"}) if apierrors.IsNotFound(err) { return nil, nil } if err != nil { return nil, err } var out []string for _, id := range items { if did := strings.TrimSpace(id.Spec.Did); did != "" && id.Spec.Mode == v1alpha1.MarketIdentityModeOAuthSessionProxy { out = append(out, did) } } sort.Strings(out) return out, nil } func (c *Composer) InfraRecordRepos(ctx context.Context, ws string) (map[string]bool, error) { ri, err := c.Client(ws, AtprotoRecordGVR) if err != nil { return nil, err } items, err := kube.List[v1alpha1.AtprotoRecord](ctx, ri, metav1.ListOptions{LabelSelector: v1alpha1.InfraKindLabel}) if apierrors.IsNotFound(err) { return map[string]bool{}, nil } if err != nil { return nil, err } out := map[string]bool{} for _, rec := range items { if rec.Spec.RepoRef != nil && strings.TrimSpace(rec.Spec.RepoRef.Name) != "" { out[strings.TrimSpace(rec.Spec.RepoRef.Name)] = true } } return out, nil } func (c *Composer) PruneSessionOwners(ctx context.Context, ws string, keep []string) error { identities, repos := map[string]bool{}, map[string]bool{} for _, did := range keep { identities[OwnerIdentityName(did)] = true repos[atprotorepo.RepoName(strings.TrimSpace(did))] = true } selector := SessionOwnerLabel + "=true" if err := c.Prune(ctx, ws, v1alpha1.AtprotoRepoGVR, selector, repos); err != nil { return err } if err := c.Prune(ctx, ws, v1alpha1.RemoteRepoGVR, selector, repos); err != nil { return err } pdses, err := c.sessionOwnerPDSes(ctx, ws, selector) if err != nil { return err } if err := c.Prune(ctx, ws, RemotePDSGVR, selector, pdses); err != nil { return err } return c.Prune(ctx, ws, MarketIdentityGVR, selector, identities) } func (c *Composer) sessionOwnerPDSes(ctx context.Context, ws, selector string) (map[string]bool, error) { ri, err := c.Client(ws, v1alpha1.RemoteRepoGVR) if err != nil { return nil, err } items, err := kube.List[v1alpha1.RemoteRepo](ctx, ri, metav1.ListOptions{LabelSelector: selector}) if apierrors.IsNotFound(err) { return map[string]bool{}, nil } if err != nil { return nil, err } out := map[string]bool{} for _, repo := range items { if repo.DeletionTimestamp == nil && strings.TrimSpace(repo.Spec.PdsRef.Name) != "" { out[strings.TrimSpace(repo.Spec.PdsRef.Name)] = true } } return out, nil } func (c *Composer) SessionOwnerAppliedPdsURL(ctx context.Context, ws, did string) (string, error) { repo, err := Get[v1alpha1.RemoteRepo](ctx, c, ws, v1alpha1.RemoteRepoGVR, RepoRefIn(did, ws).Name) if apierrors.IsNotFound(err) { return "", nil } if err != nil || repo == nil || strings.TrimSpace(repo.Spec.PdsRef.Name) == "" { return "", err } pds, err := Get[pdsv1alpha1.RemotePDS](ctx, c, ws, RemotePDSGVR, strings.TrimSpace(repo.Spec.PdsRef.Name)) if apierrors.IsNotFound(err) { return "", nil } if err != nil || pds == nil { return "", err } return strings.TrimRight(strings.TrimSpace(pds.Spec.URL), "/"), nil }