kandev/apps/backend/internal/github/webhook_service.go

package github

import (
	"context"
	"crypto/hmac"
	"crypto/sha256"
	"encoding/hex"
	"encoding/json"
	"errors"
	"fmt"
	"strings"
	"time"

	"go.uber.org/zap"

	"github.com/kandev/kandev/internal/events"
	"github.com/kandev/kandev/internal/events/bus"
)

const (
	maxGitHubWebhookPayloadSize = 10 * 1024 * 1024
	webhookDeliveryStaleAfter   = 5 * time.Minute
)

var ErrInvalidWebhookSignature = errors.New("invalid GitHub webhook signature")

type githubWebhookStore interface {
	ClaimWebhookDelivery(context.Context, *WebhookDelivery, time.Time) (WebhookDeliveryClaim, error)
	CompleteAppRegistrationWebhookDelivery(
		context.Context, string, string, WebhookDeliveryStatus, string, time.Time,
	) error
	ListWorkspaceConnectionsByAppInstallation(context.Context, string, int64) ([]*WorkspaceConnection, error)
	TransitionWorkspaceInstallationConnection(
		context.Context,
		*WorkspaceConnection,
		*WorkspaceConnection,
	) (bool, error)
	ListUserConnectionsByAppGitHubUser(context.Context, string, int64) ([]*UserConnection, error)
}

type personalConnectionRevoker interface {
	RevokePersonalConnection(context.Context, string, string) error
}

type personalAuthorizationReconciler interface {
	ReconcileAuthorizationRevocation(context.Context, *UserConnection) (bool, error)
}

type GitHubWebhookReconciliation struct {
	Installations appInstallationVerifier
	Personal      personalAuthorizationReconciler
}

type InstallationRepository struct {
	ID       int64  `json:"id"`
	FullName string `json:"full_name"`
}

type InstallationRepositoriesChange struct {
	WorkspaceID          string
	AppRegistrationID    string
	InstallationID       int64
	ConnectionSource     ConnectionSource
	CredentialGeneration int64
	Action               string
	Added                []InstallationRepository
	Removed              []InstallationRepository
}

type installationRepositoryUpdater interface {
	ApplyInstallationRepositories(context.Context, InstallationRepositoriesChange) (bool, error)
}

type GitHubWebhookRequest struct {
	DeliveryID string
	Event      string
	Signature  string
	Payload    []byte
}

func (s *Service) HandleAppRegistrationWebhook(
	ctx context.Context,
	registrationID string,
	request GitHubWebhookRequest,
) (GitHubWebhookResult, error) {
	registrationID = strings.TrimSpace(registrationID)
	runtime := s.currentAppRegistrationRuntime(registrationID)
	if registrationID == "" || runtime == nil || runtime.webhookAuth == nil ||
		runtime.webhookAuth.registrationID != registrationID {
		return GitHubWebhookResult{}, ErrGitHubNotConfigured
	}
	service := runtime.webhookAuth
	signed := service.Authenticates(request)
	result, err := service.Handle(ctx, request)
	if !signed {
		return result, err
	}
	status := DeploymentAppWebhookVerified
	reason := ""
	if err != nil {
		status = DeploymentAppWebhookFailing
		reason = "signed webhook processing failed"
	}
	observedAt := service.now().UTC()
	if s.store != nil {
		if _, healthErr := s.store.updateAppRegistrationWebhookHealth(
			ctx, registrationID, runtime.generation, status, observedAt, reason,
		); healthErr != nil && s.logger != nil {
			s.logger.Warn("App registration webhook health persistence failed", zap.Error(healthErr))
		}
	}
	return result, err
}

type GitHubWebhookResult struct {
	Duplicate bool
	Status    WebhookDeliveryStatus
	Affected  int
}

type installationWebhookEvent struct {
	Action       string `json:"action"`
	Installation struct {
		ID      int64 `json:"id"`
		Account struct {
			Login string `json:"login"`
			Type  string `json:"type"`
		} `json:"account"`
	} `json:"installation"`
}

type installationTransition struct {
	status      ConnectionStatus
	lastError   string
	login       string
	accountType string
	verified    bool
}

type GitHubWebhookService struct {
	registrationID     string
	secret             []byte
	store              githubWebhookStore
	repos              installationRepositoryUpdater
	personal           personalConnectionRevoker
	installations      appInstallationVerifier
	personalReconciler personalAuthorizationReconciler
	publisher          bus.EventBus
	now                func() time.Time
}

func NewGitHubWebhookService(
	secret string,
	store githubWebhookStore,
	repositories installationRepositoryUpdater,
	personal personalConnectionRevoker,
	publisher bus.EventBus,
	reconciliation ...GitHubWebhookReconciliation,
) *GitHubWebhookService {
	service := &GitHubWebhookService{
		secret: []byte(secret), store: store, repos: repositories, personal: personal,
		publisher: publisher, now: time.Now,
	}
	if len(reconciliation) > 0 {
		service.installations = reconciliation[0].Installations
		service.personalReconciler = reconciliation[0].Personal
	}
	return service
}

func NewAppRegistrationWebhookService(
	registrationID, secret string,
	store githubWebhookStore,
	repositories installationRepositoryUpdater,
	personal personalConnectionRevoker,
	publisher bus.EventBus,
	reconciliation ...GitHubWebhookReconciliation,
) *GitHubWebhookService {
	service := NewGitHubWebhookService(secret, store, repositories, personal, publisher, reconciliation...)
	service.registrationID = strings.TrimSpace(registrationID)
	return service
}

func (s *GitHubWebhookService) Authenticates(request GitHubWebhookRequest) bool {
	return s != nil && len(s.secret) > 0 && request.DeliveryID != "" && request.Event != "" &&
		len(request.Payload) <= maxGitHubWebhookPayloadSize &&
		validGitHubWebhookSignature(s.secret, request.Payload, request.Signature)
}

func (s *GitHubWebhookService) Handle(
	ctx context.Context,
	request GitHubWebhookRequest,
) (GitHubWebhookResult, error) {
	if s == nil || s.store == nil || len(s.secret) == 0 {
		return GitHubWebhookResult{}, errors.New("GitHub webhook service is not configured")
	}
	if request.DeliveryID == "" || request.Event == "" || len(request.Payload) > maxGitHubWebhookPayloadSize {
		return GitHubWebhookResult{}, errors.New("invalid GitHub webhook request")
	}
	if !validGitHubWebhookSignature(s.secret, request.Payload, request.Signature) {
		return GitHubWebhookResult{}, ErrInvalidWebhookSignature
	}
	now := s.now().UTC()
	claim, err := s.store.ClaimWebhookDelivery(ctx, &WebhookDelivery{
		AppRegistrationID: s.registrationID,
		DeliveryID:        request.DeliveryID,
		Event:             request.Event,
		Status:            WebhookDeliveryStatusReceived,
		ReceivedAt:        now,
	}, now.Add(-webhookDeliveryStaleAfter))
	if err != nil {
		return GitHubWebhookResult{}, fmt.Errorf("record GitHub webhook delivery: %w", err)
	}
	if !claim.Acquired {
		return GitHubWebhookResult{Duplicate: true, Status: claim.Status}, nil
	}

	affected, result, processErr := s.process(ctx, request.Event, request.Payload)
	status := WebhookDeliveryStatusProcessed
	if affected == 0 {
		status = WebhookDeliveryStatusIgnored
	}
	if processErr != nil {
		status = WebhookDeliveryStatusFailed
		result = processErr.Error()
	}
	completeErr := s.completeDelivery(ctx, request.DeliveryID, status, result, s.now().UTC())
	if processErr != nil || completeErr != nil {
		return GitHubWebhookResult{Status: status, Affected: affected}, errors.Join(processErr, completeErr)
	}
	return GitHubWebhookResult{Status: status, Affected: affected}, nil
}

func (s *GitHubWebhookService) completeDelivery(
	ctx context.Context,
	deliveryID string,
	status WebhookDeliveryStatus,
	result string,
	processedAt time.Time,
) error {
	return s.store.CompleteAppRegistrationWebhookDelivery(
		ctx, s.registrationID, deliveryID, status, result, processedAt,
	)
}

func (s *GitHubWebhookService) process(ctx context.Context, event string, payload []byte) (int, string, error) {
	switch event {
	case "installation":
		return s.processInstallation(ctx, payload)
	case "installation_repositories":
		return s.processInstallationRepositories(ctx, payload)
	case "github_app_authorization":
		return s.processAuthorization(ctx, payload)
	case "push":
		return s.processPush(ctx, payload)
	case "check_run":
		return s.processCheckRun(ctx, payload)
	default:
		return 0, "event ignored", nil
	}
}

func (s *GitHubWebhookService) processInstallation(
	ctx context.Context,
	payload []byte,
) (int, string, error) {
	var event installationWebhookEvent
	if err := json.Unmarshal(payload, &event); err != nil {
		return 0, "", fmt.Errorf("decode installation webhook: %w", err)
	}
	transition, supported, err := s.installationTransition(ctx, event)
	if err != nil {
		return 0, "", err
	}
	if !supported || event.Installation.ID <= 0 {
		return 0, "installation action ignored", nil
	}
	connections, err := s.listWorkspaceConnections(ctx, event.Installation.ID)
	if err != nil {
		return 0, "", fmt.Errorf("load installation bindings: %w", err)
	}
	affected := 0
	for _, connection := range connections {
		if !shouldApplyInstallationTransition(s.registrationID, connection, event, transition) {
			continue
		}
		updated, err := s.applyInstallationTransition(ctx, connection, transition)
		if err != nil {
			return affected, "", fmt.Errorf("update installation binding: %w", err)
		}
		if !updated {
			continue
		}
		affected++
	}
	return affected, fmt.Sprintf("installation %s: %d binding(s)", event.Action, affected), nil
}

func (s *GitHubWebhookService) listWorkspaceConnections(
	ctx context.Context,
	installationID int64,
) ([]*WorkspaceConnection, error) {
	return s.store.ListWorkspaceConnectionsByAppInstallation(ctx, s.registrationID, installationID)
}

func (s *GitHubWebhookService) installationTransition(
	ctx context.Context,
	event installationWebhookEvent,
) (installationTransition, bool, error) {
	status, lastError, supported := installationWebhookStatus(event.Action)
	transition := installationTransition{
		status: status, lastError: lastError,
		login: event.Installation.Account.Login, accountType: event.Installation.Account.Type,
	}
	if !supported || event.Installation.ID <= 0 || s.installations == nil {
		return transition, supported, nil
	}
	status, lastError, login, accountType, err := s.reconcileInstallation(ctx, event.Installation.ID)
	if err != nil {
		return installationTransition{}, false, err
	}
	return installationTransition{
		status: status, lastError: lastError, login: login, accountType: accountType, verified: true,
	}, true, nil
}

func shouldApplyInstallationTransition(
	registrationID string,
	connection *WorkspaceConnection,
	event installationWebhookEvent,
	transition installationTransition,
) bool {
	if !matchesInstallation(connection, registrationID, event.Installation.ID) {
		return false
	}
	if !transition.verified {
		return canApplyInstallationTransition(connection.Status, event.Action)
	}
	return connection.Status != transition.status ||
		(transition.login != "" && connection.InstallationAccountLogin != transition.login) ||
		(transition.accountType != "" && connection.InstallationAccountType != transition.accountType)
}

func (s *GitHubWebhookService) applyInstallationTransition(
	ctx context.Context,
	connection *WorkspaceConnection,
	transition installationTransition,
) (bool, error) {
	expected := *connection
	next := expected
	next.Status = transition.status
	next.LastError = transition.lastError
	next.CredentialGeneration++
	if transition.login != "" {
		next.InstallationAccountLogin = transition.login
	}
	if transition.accountType != "" {
		next.InstallationAccountType = transition.accountType
	}
	updated, err := s.store.TransitionWorkspaceInstallationConnection(ctx, &expected, &next)
	if updated {
		*connection = next
	}
	return updated, err
}

func (s *GitHubWebhookService) processInstallationRepositories(
	ctx context.Context,
	payload []byte,
) (int, string, error) {
	var event struct {
		Action       string `json:"action"`
		Installation struct {
			ID int64 `json:"id"`
		} `json:"installation"`
		Added   []InstallationRepository `json:"repositories_added"`
		Removed []InstallationRepository `json:"repositories_removed"`
	}
	if err := json.Unmarshal(payload, &event); err != nil {
		return 0, "", fmt.Errorf("decode installation repositories webhook: %w", err)
	}
	if event.Installation.ID <= 0 || (event.Action != "added" && event.Action != "removed") {
		return 0, "installation repositories action ignored", nil
	}
	connections, err := s.listWorkspaceConnections(ctx, event.Installation.ID)
	if err != nil {
		return 0, "", fmt.Errorf("load installation bindings: %w", err)
	}
	affected := 0
	for _, connection := range connections {
		if !matchesInstallation(connection, s.registrationID, event.Installation.ID) {
			continue
		}
		if s.repos == nil {
			return affected, "", errors.New("installation repository updater is not configured")
		}
		updated, err := s.repos.ApplyInstallationRepositories(ctx, InstallationRepositoriesChange{
			WorkspaceID:          connection.WorkspaceID,
			AppRegistrationID:    s.registrationID,
			InstallationID:       event.Installation.ID,
			ConnectionSource:     connection.Source,
			CredentialGeneration: connection.CredentialGeneration,
			Action:               event.Action,
			Added:                event.Added,
			Removed:              event.Removed,
		})
		if err != nil {
			return affected, "", fmt.Errorf("update installation repository access: %w", err)
		}
		if updated {
			affected++
		}
	}
	return affected, fmt.Sprintf("installation repositories %s: %d binding(s)", event.Action, affected), nil
}

func (s *GitHubWebhookService) processAuthorization(
	ctx context.Context,
	payload []byte,
) (int, string, error) {
	var event struct {
		Action string `json:"action"`
		Sender struct {
			ID int64 `json:"id"`
		} `json:"sender"`
	}
	if err := json.Unmarshal(payload, &event); err != nil {
		return 0, "", fmt.Errorf("decode GitHub App authorization webhook: %w", err)
	}
	if event.Action != "revoked" || event.Sender.ID <= 0 {
		return 0, "authorization action ignored", nil
	}
	connections, err := s.store.ListUserConnectionsByAppGitHubUser(ctx, s.registrationID, event.Sender.ID)
	if err != nil {
		return 0, "", fmt.Errorf("load personal GitHub bindings: %w", err)
	}
	affected := 0
	for _, connection := range connections {
		if connection == nil || connection.AppRegistrationID != s.registrationID ||
			connection.GitHubUserID != event.Sender.ID {
			continue
		}
		if s.personalReconciler != nil {
			revoked, reconcileErr := s.personalReconciler.ReconcileAuthorizationRevocation(ctx, connection)
			if reconcileErr != nil {
				return affected, "", fmt.Errorf("reconcile personal GitHub connection: %w", reconcileErr)
			}
			if revoked {
				affected++
			}
			continue
		}
		if s.personal == nil {
			return affected, "", errors.New("personal GitHub revoker is not configured")
		}
		if err := s.personal.RevokePersonalConnection(ctx, connection.WorkspaceID, connection.UserID); err != nil {
			return affected, "", fmt.Errorf("revoke personal GitHub connection: %w", err)
		}
		affected++
	}
	return affected, fmt.Sprintf("authorization revoked: %d binding(s)", affected), nil
}

type pushWebhookEvent struct {
	Ref        string `json:"ref"`
	After      string `json:"after"`
	Deleted    bool   `json:"deleted"`
	Repository struct {
		Name  string `json:"name"`
		Owner struct {
			Login string `json:"login"`
		} `json:"owner"`
	} `json:"repository"`
	Installation struct {
		ID int64 `json:"id"`
	} `json:"installation"`
	Pusher struct {
		Name string `json:"name"`
	} `json:"pusher"`
	HeadCommit struct {
		Message string `json:"message"`
	} `json:"head_commit"`
}

const gitBranchRefPrefix = "refs/heads/"

func isZeroSHA(sha string) bool {
	if sha == "" {
		return true
	}
	for _, c := range sha {
		if c != '0' {
			return false
		}
	}
	return true
}

// processPush handles the "push" webhook, publishing events.GitHubPushReceived
// with the resolved workspace IDs when the push targets a branch ref and
// isn't a branch delete.
func (s *GitHubWebhookService) processPush(ctx context.Context, payload []byte) (int, string, error) {
	var event pushWebhookEvent
	if err := json.Unmarshal(payload, &event); err != nil {
		return 0, "", fmt.Errorf("decode push webhook: %w", err)
	}
	if event.Deleted || isZeroSHA(event.After) || !strings.HasPrefix(event.Ref, gitBranchRefPrefix) {
		return 0, "push ignored", nil
	}
	workspaceIDs, err := s.resolveWorkspaceIDs(ctx, event.Installation.ID)
	if err != nil {
		return 0, "", err
	}
	if len(workspaceIDs) == 0 {
		return 0, "push ignored", nil
	}
	s.publishEvent(ctx, events.GitHubPushReceived, &GitHubPushEventPayload{
		WorkspaceIDs:  workspaceIDs,
		Owner:         event.Repository.Owner.Login,
		Name:          event.Repository.Name,
		Branch:        strings.TrimPrefix(event.Ref, gitBranchRefPrefix),
		SHA:           event.After,
		PusherLogin:   event.Pusher.Name,
		HeadCommitMsg: event.HeadCommit.Message,
	})
	return len(workspaceIDs), fmt.Sprintf("push published: %d workspace(s)", len(workspaceIDs)), nil
}

type checkRunWebhookEvent struct {
	Action   string `json:"action"`
	CheckRun struct {
		ID         int64  `json:"id"`
		Name       string `json:"name"`
		HeadSHA    string `json:"head_sha"`
		Conclusion string `json:"conclusion"`
		HTMLURL    string `json:"html_url"`
		CheckSuite struct {
			HeadBranch string `json:"head_branch"`
		} `json:"check_suite"`
	} `json:"check_run"`
	Repository struct {
		Name  string `json:"name"`
		Owner struct {
			Login string `json:"login"`
		} `json:"owner"`
	} `json:"repository"`
	Installation struct {
		ID int64 `json:"id"`
	} `json:"installation"`
}

// processCheckRun handles the "check_run" webhook, publishing
// events.GitHubCheckRunCompleted with the resolved workspace IDs for
// completed check runs only.
func (s *GitHubWebhookService) processCheckRun(ctx context.Context, payload []byte) (int, string, error) {
	var event checkRunWebhookEvent
	if err := json.Unmarshal(payload, &event); err != nil {
		return 0, "", fmt.Errorf("decode check_run webhook: %w", err)
	}
	if event.Action != "completed" {
		return 0, "check_run ignored", nil
	}
	workspaceIDs, err := s.resolveWorkspaceIDs(ctx, event.Installation.ID)
	if err != nil {
		return 0, "", err
	}
	if len(workspaceIDs) == 0 {
		return 0, "check_run ignored", nil
	}
	s.publishEvent(ctx, events.GitHubCheckRunCompleted, &GitHubCheckRunEventPayload{
		WorkspaceIDs: workspaceIDs,
		Owner:        event.Repository.Owner.Login,
		Name:         event.Repository.Name,
		Branch:       event.CheckRun.CheckSuite.HeadBranch,
		SHA:          event.CheckRun.HeadSHA,
		CheckName:    event.CheckRun.Name,
		Conclusion:   event.CheckRun.Conclusion,
		CheckRunID:   event.CheckRun.ID,
		HTMLURL:      event.CheckRun.HTMLURL,
	})
	return len(workspaceIDs), fmt.Sprintf("check_run published: %d workspace(s)", len(workspaceIDs)), nil
}

// resolveWorkspaceIDs resolves an installation ID to the distinct set of
// workspace IDs connected to it under this registration.
func (s *GitHubWebhookService) resolveWorkspaceIDs(ctx context.Context, installationID int64) ([]string, error) {
	if installationID <= 0 {
		return nil, nil
	}
	connections, err := s.listWorkspaceConnections(ctx, installationID)
	if err != nil {
		return nil, fmt.Errorf("load installation bindings: %w", err)
	}
	seen := make(map[string]bool, len(connections))
	workspaceIDs := make([]string, 0, len(connections))
	for _, connection := range connections {
		// Only active installations may drive automation events: a suspended
		// or revoked App binding must not still fire push/CI automations.
		if connection == nil || connection.WorkspaceID == "" ||
			connection.Status != ConnectionStatusActive || seen[connection.WorkspaceID] {
			continue
		}
		seen[connection.WorkspaceID] = true
		workspaceIDs = append(workspaceIDs, connection.WorkspaceID)
	}
	return workspaceIDs, nil
}

// publishEvent is nil-safe and never fails webhook delivery: publish errors
// are swallowed so a bus outage doesn't turn a real webhook into a failed
// delivery.
func (s *GitHubWebhookService) publishEvent(ctx context.Context, eventType string, data interface{}) {
	if s.publisher == nil {
		return
	}
	_ = s.publisher.Publish(ctx, eventType, bus.NewEvent(eventType, "github_webhook_service", data))
}

func (s *GitHubWebhookService) reconcileInstallation(
	ctx context.Context,
	installationID int64,
) (ConnectionStatus, string, string, string, error) {
	installation, err := s.installations.GetInstallation(ctx, installationID)
	if err != nil {
		var apiErr *GitHubAPIError
		if errors.As(err, &apiErr) && apiErr.StatusCode == 404 {
			return ConnectionStatusRevoked, "GitHub App installation deleted", "", "", nil
		}
		return "", "", "", "", fmt.Errorf("reconcile GitHub App installation: %w", err)
	}
	if installation.ID != installationID {
		return "", "", "", "", errors.New("reconciled GitHub App installation ID does not match webhook")
	}
	if installation.SuspendedAt != nil {
		return ConnectionStatusSuspended, "GitHub App installation suspended",
			installation.AccountLogin, installation.AccountType, nil
	}
	return ConnectionStatusActive, "", installation.AccountLogin, installation.AccountType, nil
}

func validGitHubWebhookSignature(secret, payload []byte, provided string) bool {
	if !strings.HasPrefix(provided, "sha256=") {
		return false
	}
	providedMAC, err := hex.DecodeString(strings.TrimPrefix(provided, "sha256="))
	if err != nil {
		return false
	}
	mac := hmac.New(sha256.New, secret)
	_, _ = mac.Write(payload)
	return hmac.Equal(mac.Sum(nil), providedMAC)
}

func installationWebhookStatus(action string) (ConnectionStatus, string, bool) {
	switch action {
	case "suspend":
		return ConnectionStatusSuspended, "GitHub App installation suspended", true
	case "unsuspend":
		return ConnectionStatusActive, "", true
	case "deleted":
		return ConnectionStatusRevoked, "GitHub App installation deleted", true
	default:
		return "", "", false
	}
}

func canApplyInstallationTransition(current ConnectionStatus, action string) bool {
	switch action {
	case "suspend":
		return current != ConnectionStatusSuspended && current != ConnectionStatusRevoked
	case "unsuspend":
		return current == ConnectionStatusSuspended || current == ConnectionStatusInvalid
	case "deleted":
		return current != ConnectionStatusRevoked
	default:
		return false
	}
}

func matchesInstallation(connection *WorkspaceConnection, registrationID string, installationID int64) bool {
	return connection != nil && connection.Source == ConnectionSourceGitHubAppInstallation &&
		connection.AppRegistrationID == registrationID &&
		connection.InstallationID != nil && *connection.InstallationID == installationID
}

func (s *Store) updateAppRegistrationWebhookHealth(
	ctx context.Context,
	registrationID string,
	generation int64,
	status DeploymentAppWebhookStatus,
	lastWebhookAt time.Time,
	lastError string,
) (bool, error) {
	result, err := s.db.ExecContext(ctx, s.db.Rebind(`
		UPDATE github_app_registrations
		SET webhook_status = ?, last_webhook_at = ?, last_error = ?, updated_at = ?
		WHERE id = ? AND credential_generation = ?`),
		status, lastWebhookAt, nullString(lastError), time.Now().UTC(), registrationID, generation,
	)
	if err != nil {
		return false, err
	}
	count, err := result.RowsAffected()
	return count == 1, err
}