28449634629

Committed 30 Jun 2026 01:53PM UTC coverage: 62.788% (+0.04%) from 62.75%

Build # 28449634629

Build Type

Pull #565

github

Committed by

emmdim

Commit Message

fix(api): enforce integrator process quota on the /transactions path

PR #557 added an `if !managed(org)` skip in HasTxPermission, deferring the
managed-org process-count check to the integrator's aggregate ManagedProcesses
quota (ReserveManagedPublish) "at publish time". But that reservation was only
wired into publishProcessHandler (/process/{id}/publish); the remote-signer
/transactions path (signTxHandler) — the primary SDK flow — was left calling
HasTxPermission with the check skipped and never reserving.

Result: managed orgs could mint unlimited non-test elections via /transactions,
bypassing the integrator cap, and /integrator under-reported ManagedProcesses.

Extract the reserve logic from publishProcessHandler into a shared
reserveManagedProcessSlot helper and call it from both NEW_PROCESS entry points
so they cannot drift apart again. signTxHandler rolls the reservation back if
signing fails, mirroring the publish path.

Adds TestIntegratorProcessQuotaViaTransactions covering the counter increment,
the aggregate cap, and the test-sized exemption on /transactions.

Pull Request Pull Request #565: fix(api): enforce integrator process quota on the /transactions path

Coverage Stats

36 of 52 new or added lines in 2 files covered. (69.23%)

10301 of 16406 relevant lines covered (62.79%)

44.98 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

56.13

/api/process.go

package api

import (
        "context"
        "encoding/json"
        "io"
        "net/http"
        "strings"
        "time"

        "github.com/ethereum/go-ethereum/common"
        "github.com/go-chi/chi/v5"
        "github.com/vocdoni/saas-backend/account"
        "github.com/vocdoni/saas-backend/api/apicommon"
        "github.com/vocdoni/saas-backend/db"
        "github.com/vocdoni/saas-backend/errors"
        "github.com/vocdoni/saas-backend/internal"
        "github.com/vocdoni/saas-backend/subscriptions"
        "go.mongodb.org/mongo-driver/bson/primitive"
        dvoteapi "go.vocdoni.io/dvote/api"
        "go.vocdoni.io/dvote/log"
        "go.vocdoni.io/proto/build/go/models"
)

// createProcessHandler godoc
//
//        @Summary                Create a new voting process
//        @Description        Create a new voting process (draft). Requires Manager/Admin role of the owning organization. The
//        @Description        request must reference the organization via either a `censusId` or an explicit `orgAddress`.
//        @Description        Creating a draft consumes the plan's draft permission.
//        @Description
//        @Description        The optional `electionParams` field carries the on-chain election definition (questions, dates,
//        @Description        vote/election types) used later to publish the draft on chain via POST /process/{processId}/publish.
//        @Description        It is additive: omitting it keeps the legacy draft behavior.
//        @Description
//        @Description        Also callable with a scoped API key (scope: `voting:write`).
//        @Tags                        process
//        @Accept                        json
//        @Produce                json
//        @Security                BearerAuth
//        @Param                        request        body                apicommon.CreateProcessRequest        true        "Process creation information (optionally with electionParams)"
//        @Success                200                {object}        primitive.ObjectID                                "Bare JSON string: 24-hex draft ProcessID, e.g. 507f1f77bcf86cd799439011"
//        @Failure                400                {object}        errors.Error                                        "Invalid input data"
//        @Failure                401                {object}        errors.Error                                        "Unauthorized"
//        @Failure                403                {object}        errors.Error                                        "Plan does not allow creating more drafts"
//        @Failure                404                {object}        errors.Error                                        "Published census not found"
//        @Failure                500                {object}        errors.Error                                        "Internal server error"
//        @Router                        /process [post]
func (a *API) createProcessHandler(w http.ResponseWriter, r *http.Request) {
        // parse the process info from the request body
        processInfo := &apicommon.CreateProcessRequest{}
        if err := json.NewDecoder(r.Body).Decode(&processInfo); err != nil {
                errors.ErrMalformedBody.Write(w)
                return
        }

        // get the user from the request context
        user, ok := apicommon.UserFromContext(r.Context())
        if !ok {
                errors.ErrUnauthorized.Write(w)
                return
        }

        // if it's a draft process
        if processInfo.Address.Equals(nil) && processInfo.OrgAddress == (common.Address{}) {
                errors.ErrMalformedBody.Withf("draft processes must provide an org address").Write(w)
                return
        }

        // Create or update the process
        process := &db.Process{
                Metadata:       processInfo.Metadata,
                ElectionParams: processInfo.ElectionParams,
        }

        var orgAddress common.Address
        if processInfo.CensusID != nil {
                var err error
                census, err := a.db.Census(processInfo.CensusID.String())
                if err != nil {
                        if err == db.ErrNotFound {
                                errors.ErrMalformedURLParam.Withf("invalid census provided").Write(w)
                                return
                        }
                        errors.ErrGenericInternalServerError.WithErr(err).Write(w)
                        return
                }
                process.Census = *census
                orgAddress = census.OrgAddress
        } else if processInfo.OrgAddress != (common.Address{}) {
                orgAddress = processInfo.OrgAddress
        } else {
                errors.ErrMalformedBody.Withf("either census ID or organization address must be provided").Write(w)
                return
        }

        // check the user has the necessary permissions
        if !user.HasRoleFor(orgAddress, db.ManagerRole) && !user.HasRoleFor(orgAddress, db.AdminRole) {
                errors.ErrUnauthorized.Withf("user is not admin of organization").Write(w)
                return
        }

        process.OrgAddress = orgAddress
        if !processInfo.Address.Equals(nil) {
                process.Address = processInfo.Address
        }

        // if it's a new draft process
        if process.Address.Equals(nil) && process.ID == primitive.NilObjectID {
                if err := a.subscriptions.OrgHasPermission(process.OrgAddress, subscriptions.CreateDraft); err != nil {
                        if apierr, ok := err.(errors.Error); ok {
                                apierr.Write(w)
                                return
                        }
                        errors.ErrGenericInternalServerError.WithErr(err).Write(w)
                        return
                }
        }

        processID, err := a.db.SetProcess(process)
        if err != nil {
                errors.ErrGenericInternalServerError.WithErr(err).Write(w)
                return
        }

        apicommon.HTTPWriteJSON(w, processID)
}

// updateProcessHandler godoc
//
//        @Summary                Update an existing voting process
//        @Description        Update an existing draft voting process. Requires Manager/Admin role of the owning organization.
//        @Description        Only drafts that have not yet been published on chain can be updated; updating a published process
//        @Description        returns a 409. The optional `electionParams` field can be set or replaced here (additive), same as on
//        @Description        create.
//        @Tags                        process
//        @Accept                        json
//        @Produce                json
//        @Security                BearerAuth
//        @Param                        processId        path                string                                                        true        "24-hex draft ProcessID"
//        @Param                        request                body                apicommon.UpdateProcessRequest        true        "Process update information (optionally with electionParams)"
//        @Success                200                        {string}        string                                                        "OK"
//        @Failure                400                        {object}        errors.Error                                        "Invalid input data"
//        @Failure                401                        {object}        errors.Error                                        "Unauthorized"
//        @Failure                404                        {object}        errors.Error                                        "Process not found"
//        @Failure                409                        {object}        errors.Error                                        "Process already published and not in draft mode"
//        @Failure                500                        {object}        errors.Error                                        "Internal server error"
//        @Router                        /process/{processId} [put]
func (a *API) updateProcessHandler(w http.ResponseWriter, r *http.Request) {
        processID := chi.URLParam(r, "processId")
        if processID == "" {
                errors.ErrMalformedURLParam.Withf("missing process ID").Write(w)
                return
        }
        parsedID, err := primitive.ObjectIDFromHex(processID)
        if err != nil {
                errors.ErrMalformedURLParam.Withf("invalid process ID").Write(w)
                return
        }

        // parse the process info from the request body
        processInfo := &apicommon.UpdateProcessRequest{}
        if err := json.NewDecoder(r.Body).Decode(&processInfo); err != nil {
                errors.ErrMalformedBody.Write(w)
                return
        }

        // get the user from the request context
        user, ok := apicommon.UserFromContext(r.Context())
        if !ok {
                errors.ErrUnauthorized.Write(w)
                return
        }

        existingProcess, err := a.db.Process(parsedID)
        if err != nil {
                if err == db.ErrNotFound {
                        errors.ErrProcessNotFound.Write(w)
                        return
                }
                errors.ErrGenericInternalServerError.WithErr(err).Write(w)
                return
        }

        // check if it's a draft process and can be overwritten
        if !existingProcess.Address.Equals(nil) {
                errors.ErrDuplicateConflict.Withf("process already exists and is not in draft mode").Write(w)
                return
        }

        // check the user has the necessary permissions
        if !user.HasRoleFor(existingProcess.OrgAddress, db.ManagerRole) &&
                !user.HasRoleFor(existingProcess.OrgAddress, db.AdminRole) {
                errors.ErrUnauthorized.Withf("user is not admin or manager of the organization that owns this process").Write(w)
                return
        }

        var census *db.Census
        if !processInfo.CensusID.Equals(nil) {
                census, err = a.db.Census(processInfo.CensusID.String())
                if err != nil {
                        if err == db.ErrNotFound {
                                errors.ErrMalformedURLParam.Withf("census not found").Write(w)
                                return
                        }
                        errors.ErrGenericInternalServerError.WithErr(err).Write(w)
                        return
                }
                existingProcess.Census = *census
        }

        if len(processInfo.Metadata) > 0 {
                existingProcess.Metadata = processInfo.Metadata
        }

        if processInfo.ElectionParams != nil {
                existingProcess.ElectionParams = processInfo.ElectionParams
        }

        if !processInfo.Address.Equals(nil) {
                existingProcess.Address = processInfo.Address
        }

        _, err = a.db.SetProcess(existingProcess)
        if err != nil {
                errors.ErrGenericInternalServerError.WithErr(err).Write(w)
                return
        }

        apicommon.HTTPWriteOK(w)
}

// processInfoHandler godoc
//
//        @Summary                Get process information
//        @Description        Retrieve voting process information by ID. Returns process details including census and metadata.
//        @Description        For encrypted (secretUntilTheEnd) elections the response additionally carries the election's
//        @Description        encryption public keys in `encryptionKeys` (one per keykeeper) once they are published on chain;
//        @Description        the field is absent for non-encrypted elections and before the keys are published.
//        @Tags                        process
//        @Accept                        json
//        @Produce                json
//        @Param                        processId        path                string        true        "Process ID"
//        @Success                200                        {object}        apicommon.ProcessInfo
//        @Failure                400                        {object}        errors.Error        "Invalid process ID"
//        @Failure                404                        {object}        errors.Error        "Process not found"
//        @Failure                500                        {object}        errors.Error        "Internal server error"
//        @Router                        /process/{processId} [get]
func (a *API) processInfoHandler(w http.ResponseWriter, r *http.Request) {
        processID := chi.URLParam(r, "processId")
        if len(processID) == 0 {
                errors.ErrMalformedURLParam.Withf("missing process ID").Write(w)
                return
        }
        parsedID, err := primitive.ObjectIDFromHex(processID)
        if err != nil {
                errors.ErrMalformedURLParam.Withf("invalid process ID").Write(w)
                return
        }

        process, err := a.db.Process(parsedID)
        if err != nil {
                if err == db.ErrNotFound {
                        errors.ErrProcessNotFound.Write(w)
                        return
                }
                errors.ErrGenericInternalServerError.WithErr(err).Write(w)
                return
        }

        // resolve the canonical metadata from its reference (best-effort): the Vochain only
        // resolves ipfs:// pointers, so the https /storage pointer this service publishes is
        // resolved here instead.
        if md := a.resolveProcessMetadata(r.Context(), process); md != nil {
                process.Metadata = md
        }

        // for encrypted (secretUntilTheEnd) elections, expose the encryption public keys the
        // voter needs to encrypt their ballot. Gated and cached so the common (non-encrypted)
        // read path never touches the chain.
        process.EncryptionKeys = a.resolveProcessEncryptionKeys(process)

        apicommon.HTTPWriteJSON(w, &apicommon.ProcessInfo{
                Process: process,
                ChainID: a.account.ChainID(),
        })
}

// resolveProcessEncryptionKeys returns the encryption public keys for an encrypted
// election, fetching them from the Vochain only when needed and caching them on the
// process once published. It is a no-op (returns nil, no chain call) for non-encrypted
// elections, unpublished drafts, and processes whose keys are already cached are served
// from the cache. The result is cached only when non-empty: between an election's creation
// and the keykeepers publishing its keys the node returns an empty set, and a later read
// must still pick them up.
func (a *API) resolveProcessEncryptionKeys(p *db.Process) []db.EncryptionKey {
        // gate: only encrypted elections have encryption keys. Reading the stored election type
        // keeps every other process on a chain-free path.
        if p.ElectionParams == nil || !p.ElectionParams.ElectionType.SecretUntilTheEnd {
                return nil
        }
        // nothing on chain yet (draft) — no keys to fetch
        if p.Address.Equals(nil) {
                return nil
        }
        // immutable once published: serve the cached keys without a chain round-trip
        if len(p.EncryptionKeys) > 0 {
                return p.EncryptionKeys
        }
        keys, err := a.account.ElectionEncryptionKeys(p.Address.Bytes())
        if err != nil {
                log.Warnw("encryption keys: election keys fetch failed", "process", p.Address.String(), "error", err)
                return nil
        }
        // not published yet — do not cache an empty set so a later read still resolves them
        if len(keys) == 0 {
                return nil
        }
        if err := a.db.SetProcessEncryptionKeys(p.ID, keys); err != nil {
                log.Warnw("encryption keys: could not persist keys", "process", p.Address.String(), "error", err)
        }
        return keys
}

// metadataObjectUserID is the object-storage owner recorded for server-generated
// metadata documents cached on read.
const metadataObjectUserID = "system"

// resolveProcessMetadata returns the canonical metadata for a published process,
// consulting process.MetadataURL. Returns nil (caller keeps the stored value) for
// drafts or on any error. Persists the reference inline only when it changed.
func (a *API) resolveProcessMetadata(ctx context.Context, p *db.Process) map[string]any {
        if p.Address.Equals(nil) { // draft, nothing on chain
                return nil
        }
        var election *dvoteapi.Election
        changed := false
        // bootstrap the reference from the on-chain pointer if we don't have one yet
        if p.MetadataURL == "" {
                el, err := a.account.Election(p.Address.Bytes())
                if err != nil {
                        log.Warnw("metadata: election fetch failed", "process", p.Address.String(), "error", err)
                        return nil
                }
                election, p.MetadataURL, changed = el, el.MetadataURL, true
        }

        // resolve the document. Branches set m on success and leave it nil on failure; no
        // early return, so a bootstrapped reference is still persisted below (avoids
        // re-deriving it from the chain on every read).
        var m map[string]any
        name, isLocal := a.objectStorage.LocalName(p.MetadataURL)
        switch {
        case isLocal:
                // our own object storage — resolve locally; treat a corrupt/non-JSON object as a
                // resolution failure (m stays nil) so the stored metadata is kept.
                if obj, err := a.objectStorage.GetByName(name); err == nil {
                        if json.Unmarshal(obj.Data, &m) != nil {
                                m = nil
                        }
                }
        case strings.HasPrefix(p.MetadataURL, "ipfs://"):
                // the Vochain resolves ipfs; reuse the election if already fetched
                if election == nil {
                        if el, err := a.account.Election(p.Address.Bytes()); err == nil {
                                election = el
                        }
                }
                if election != nil {
                        if mm, ok := election.Metadata.(map[string]any); ok && len(mm) > 0 {
                                m = mm
                        }
                }
        case strings.HasPrefix(p.MetadataURL, "http://"), strings.HasPrefix(p.MetadataURL, "https://"):
                m = fetchExternalMetadata(ctx, p.MetadataURL)
        default:
                // unrecognized scheme — leave m nil; a bootstrapped reference is still persisted below
        }

        // metadata is immutable (a change is a republish through this API), so cache any
        // remotely-resolved document locally and promote the reference — later reads then
        // resolve from our own storage with no Vochain or external round-trip.
        if m != nil && !isLocal {
                if b, err := json.Marshal(m); err == nil {
                        if stored, err := a.objectStorage.PutJSON(b, metadataObjectUserID); err == nil {
                                p.MetadataURL, changed = a.objectStorage.LocalURL(stored), true
                        }
                }
        }
        // persist a learned/promoted reference with a targeted update (not a full rewrite) so
        // a concurrent change to other process fields is not clobbered by this read path. This
        // runs even when resolution failed, so a bootstrapped reference is stored once.
        if changed {
                if err := a.db.SetProcessMetadataURL(p.ID, p.MetadataURL); err != nil {
                        log.Warnw("metadata: could not persist metadataURL", "process", p.Address.String(), "error", err)
                }
        }
        return m
}

// fetchExternalMetadata best-effort downloads and JSON-decodes a metadata document from
// an external http(s) reference (one that does not point at our own object storage).
// Returns nil on any failure.
func fetchExternalMetadata(ctx context.Context, url string) map[string]any {
        ctx, cancel := context.WithTimeout(ctx, 3*time.Second)
        defer cancel()
        req, err := http.NewRequestWithContext(ctx, http.MethodGet, url, nil)
        if err != nil {
                return nil
        }
        resp, err := http.DefaultClient.Do(req)
        if err != nil {
                return nil
        }
        defer func() { _ = resp.Body.Close() }()
        if resp.StatusCode != http.StatusOK {
                return nil
        }
        var m map[string]any
        if json.NewDecoder(io.LimitReader(resp.Body, 1<<20)).Decode(&m) != nil { // 1 MiB cap
                return nil
        }
        return m
}

// organizationListProcessDraftsHandler godoc
//
//        @Summary                Get paginated list of process drafts
//        @Description        Returns a list of voting process drafts.
//        @Tags                        process
//        @Accept                        json
//        @Produce                json
//        @Success                200        {object}        apicommon.ListOrganizationProcesses
//        @Failure                404        {object}        errors.Error        "Process not found"
//        @Failure                500        {object}        errors.Error        "Internal server error"
//        @Router                        /organizations/{address}/processes/drafts [get]
func (a *API) organizationListProcessDraftsHandler(w http.ResponseWriter, r *http.Request) {
        // get the organization info from the request context
        org, _, ok := a.organizationFromRequest(r)
        if !ok {
                errors.ErrNoOrganizationProvided.Write(w)
                return
        }
        // get the user from the request context
        user, ok := apicommon.UserFromContext(r.Context())
        if !ok {
                errors.ErrUnauthorized.Write(w)
                return
        }
        // check the user has the necessary permissions
        if !user.HasRoleFor(org.Address, db.ManagerRole) && !user.HasRoleFor(org.Address, db.AdminRole) {
                errors.ErrUnauthorized.Withf("user is not admin of organization").Write(w)
                return
        }

        params, err := parsePaginationParams(r.URL.Query().Get(ParamPage), r.URL.Query().Get(ParamLimit))
        if err != nil {
                errors.ErrMalformedURLParam.WithErr(err).Write(w)
                return
        }

        // retrieve the orgMembers with pagination
        totalItems, processes, err := a.db.ListProcesses(org.Address, params.Page, params.Limit, db.DraftOnly)
        if err != nil {
                errors.ErrGenericInternalServerError.Withf("could not get processes: %v", err).Write(w)
                return
        }
        pagination, err := calculatePagination(params.Page, params.Limit, totalItems)
        if err != nil {
                errors.ErrMalformedURLParam.WithErr(err).Write(w)
                return
        }

        apicommon.HTTPWriteJSON(w, &apicommon.ListOrganizationProcesses{
                Pagination: pagination,
                Processes:  processes,
        })
}

// deleteProcessHandler godoc
//
//        @Summary                Delete a draft voting process
//        @Description        Delete a draft voting process. Only unpublished drafts can be deleted; a process
//        @Description        already published on-chain cannot be removed. Requires Manager/Admin role of the
//        @Description        organization that owns the process.
//        @Description
//        @Description        Also callable with a scoped API key (scope: `voting:write`).
//        @Tags                        process
//        @Accept                        json
//        @Produce                json
//        @Security                BearerAuth
//        @Param                        processId        path                string                        true        "Process ID"
//        @Success                200                        {string}        string                        "OK"
//        @Failure                400                        {object}        errors.Error        "Invalid process ID"
//        @Failure                401                        {object}        errors.Error        "Unauthorized"
//        @Failure                404                        {object}        errors.Error        "Process not found"
//        @Failure                409                        {object}        errors.Error        "Process already published and not in draft mode"
//        @Failure                500                        {object}        errors.Error        "Internal server error"
//        @Router                        /process/{processId} [delete]
func (a *API) deleteProcessHandler(w http.ResponseWriter, r *http.Request) {
        processID := chi.URLParam(r, "processId")
        if processID == "" {
                errors.ErrMalformedURLParam.Withf("missing process ID").Write(w)
                return
        }
        parsedID, err := primitive.ObjectIDFromHex(processID)
        if err != nil {
                errors.ErrMalformedURLParam.Withf("invalid process ID").Write(w)
                return
        }

        // get the user from the request context
        user, ok := apicommon.UserFromContext(r.Context())
        if !ok {
                errors.ErrUnauthorized.Write(w)
                return
        }

        existingProcess, err := a.db.Process(parsedID)
        if err != nil {
                if err == db.ErrNotFound {
                        errors.ErrProcessNotFound.Withf("process not found").Write(w)
                        return
                }
                errors.ErrGenericInternalServerError.WithErr(err).Write(w)
                return
        }

        // only draft processes can be deleted; a published process has a non-nil
        // on-chain address and lives on the Vochain, where it cannot be removed.
        if !existingProcess.Address.Equals(nil) {
                errors.ErrDuplicateConflict.Withf("process already published and not in draft mode").Write(w)
                return
        }

        // check the user has the necessary permissions
        if !user.HasRoleFor(existingProcess.OrgAddress, db.ManagerRole) &&
                !user.HasRoleFor(existingProcess.OrgAddress, db.AdminRole) {
                errors.ErrUnauthorized.Withf("user is not admin or manager of the organization that owns this process").Write(w)
                return
        }

        err = a.db.DelProcess(parsedID)
        if err != nil {
                errors.ErrGenericInternalServerError.WithErr(err).Write(w)
                return
        }

        apicommon.HTTPWriteOK(w)
}

// reserveManagedProcessSlot reserves one process slot against the integrator's shared
// ManagedProcesses quota when org is a managed organization publishing a non-test-sized
// election. It returns the integrator address and reserved=true when a slot was taken — the
// caller MUST roll it back with AddOrganizationManagedProcesses(integratorAddr, -1) if the
// publish/sign later fails. For a standalone org or a test-sized election it is a no-op
// (reserved=false, nil error).
//
// HasTxPermission skips the per-org process-count check for managed orgs precisely because
// this integrator-level reservation enforces it instead; every NEW_PROCESS entry point (draft
// publish and the remote-signer /transactions path) must call this so the two cannot drift.
func (a *API) reserveManagedProcessSlot(org *db.Organization, maxCensusSize uint64) (common.Address, bool, error) {
        if org.ManagedBy == (common.Address{}) || maxCensusSize <= uint64(db.TestMaxCensusSize) {
                return common.Address{}, false, nil
        }
        integrator, err := a.db.Organization(org.ManagedBy)
        if err != nil {
                // a managed org whose integrator no longer exists is a not-found condition, not a
                // server fault — map it like subscriptions.limitsOwner does rather than 500.
                if err == db.ErrNotFound {
                        return common.Address{}, false, errors.ErrOrganizationNotFound.WithErr(err)
                }
                return common.Address{}, false, errors.ErrGenericInternalServerError.Withf("could not get integrator organization: %v", err)
        }
        maxProcesses, err := a.subscriptions.ManagedPublishLimits(integrator)
        if err != nil {
                return common.Address{}, false, err
        }
        if err := a.db.ReserveManagedPublish(integrator.Address, maxProcesses); err != nil {
                if err == db.ErrManagedQuotaReached {
                        return common.Address{}, false, errors.ErrIntegratorQuotaExceeded
                }
                return common.Address{}, false, errors.ErrGenericInternalServerError.WithErr(err)
        }
        return integrator.Address, true, nil
}

// publishProcessHandler godoc
//
//        @Summary                Publish a draft process as an on-chain election
//        @Description        Publishes an existing draft process (created via POST /process) as an on-chain
//        @Description        election. The backend builds the election metadata and NewProcess transaction,
//        @Description        funds and signs it (with the organization signer) synchronously, then submits and
//        @Description        confirms it on a background worker; the call returns 202 with a job id. Poll GET
//        @Description        /jobs/{jobId} for the resulting on-chain id. Requires Admin role. Idempotent: if
//        @Description        the draft is already published its on-chain id is returned with 200 without sending
//        @Description        a new transaction. Publishing under a managed organization additionally enforces the
//        @Description        integrator's aggregate election (process-count) quota.
//        @Description
//        @Description        Also callable with a scoped API key (scope: `voting:write`).
//        @Tags                        process
//        @Accept                        json
//        @Produce                json
//        @Security                BearerAuth
//        @Param                        processId        path                string                                                                true        "Draft process ID"
//        @Success                202                        {object}        apicommon.EnqueuedResponse                        "Publish accepted; poll GET /jobs/{jobId}"
//        @Success                200                        {object}        apicommon.PublishProcessResponse        "Already published (idempotent): on-chain id and status"
//        @Failure                400                        {object}        errors.Error                                                "Invalid input data"
//        @Failure                401                        {object}        errors.Error                                                "Unauthorized"
//        @Failure                404                        {object}        errors.Error                                                "Process not found"
//        @Failure                409                        {object}        errors.Error                                                "A publish is already in progress for this draft"
//        @Failure                500                        {object}        errors.Error                                                "Internal server error"
//        @Failure                503                        {object}        errors.Error                                                "Transaction queue is full"
//        @Router                        /process/{processId}/publish [post]
func (a *API) publishProcessHandler(w http.ResponseWriter, r *http.Request) {
        processID := chi.URLParam(r, "processId")
        if processID == "" {
                errors.ErrMalformedURLParam.Withf("missing process ID").Write(w)
                return
        }
        parsedID, err := primitive.ObjectIDFromHex(processID)
        if err != nil {
                errors.ErrMalformedURLParam.Withf("invalid process ID").Write(w)
                return
        }

        user, ok := apicommon.UserFromContext(r.Context())
        if !ok {
                errors.ErrUnauthorized.Write(w)
                return
        }

        draft, err := a.db.Process(parsedID)
        if err != nil {
                if err == db.ErrNotFound {
                        errors.ErrProcessNotFound.Write(w)
                        return
                }
                errors.ErrGenericInternalServerError.WithErr(err).Write(w)
                return
        }

        // permission: Admin of the owning organization. Publishing maps to a NEW_PROCESS
        // tx, which subscriptions.HasTxPermission requires Admin for, so we enforce the
        // same role here rather than letting a Manager pass and be rejected downstream.
        if !user.HasRoleFor(draft.OrgAddress, db.AdminRole) {
                errors.ErrUnauthorized.Withf("user is not admin of the organization that owns this process").Write(w)
                return
        }

        // idempotent: already published
        if !draft.Address.Equals(nil) {
                apicommon.HTTPWriteJSON(w, &apicommon.PublishProcessResponse{Address: draft.Address, Status: draft.Status})
                return
        }

        // a draft must carry the high-level election definition to be publishable.
        // the on-chain census root is always the CSP public key, so the draft census
        // (member list used later by the CSP bundle flow) is not required here.
        if draft.ElectionParams == nil {
                errors.ErrMalformedBody.Withf("draft has no election params").Write(w)
                return
        }

        // atomically claim the draft for publishing BEFORE any expensive metadata/funding/
        // signing work. This single conditional update is the authoritative duplicate-publish
        // guard: two concurrent publishes cannot both win the claim, so only one NEW_PROCESS
        // tx is ever built for a draft.
        claimed, err := a.db.ClaimProcessForPublish(parsedID)
        if err != nil {
                errors.ErrGenericInternalServerError.WithErr(err).Write(w)
                return
        }
        if !claimed {
                // lost the claim: either another request is mid-publish, or it already finished.
                if cur, e := a.db.Process(parsedID); e == nil && !cur.Address.Equals(nil) {
                        apicommon.HTTPWriteJSON(w, &apicommon.PublishProcessResponse{Address: cur.Address, Status: cur.Status})
                        return
                }
                errors.ErrPublishInProgress.Write(w)
                return
        }

        // from here the draft is in PUBLISHING. Until a worker owns the job, every failure
        // path must release the claim, otherwise the draft is stuck (the release must $unset
        // status, since a zero-value string write is dropped by the dynamic update helper).
        committed := false
        defer func() {
                if committed {
                        return
                }
                if e := a.db.ClearProcessPublishing(parsedID); e != nil {
                        log.Warnw("could not clear publishing state after failed publish", "error", e)
                }
        }()

        org, err := a.db.Organization(draft.OrgAddress)
        if err != nil {
                if err == db.ErrNotFound {
                        errors.ErrOrganizationNotFound.Write(w)
                        return
                }
                errors.ErrGenericInternalServerError.WithErr(err).Write(w)
                return
        }

        // if this org is managed by an integrator, atomically reserve the integrator's
        // aggregate process quota BEFORE building the on-chain tx, so concurrent publishes
        // cannot each pass a stale check and exceed the cap. The reservation is rolled back
        // (deferred) unless the publish commits. Test-sized elections are exempt from the
        // integrator quota, mirroring the per-org Processes counter exemption below.
        nonTestSized := draft.ElectionParams.MaxCensusSize > uint64(db.TestMaxCensusSize)
        integratorAddr, managedReserved, err := a.reserveManagedProcessSlot(org, draft.ElectionParams.MaxCensusSize)
        if err != nil {
                if apiErr, ok := err.(errors.Error); ok {
                        apiErr.Write(w)
                        return
                }
                errors.ErrGenericInternalServerError.WithErr(err).Write(w)
                return
        }
        if managedReserved {
                // roll back the reservation if the publish is not handed to a worker (any
                // synchronous failure below, or a full queue). Once enqueued the worker owns
                // the reservation outcome and clears managedReserved.
                defer func() {
                        if !managedReserved {
                                return
                        }
                        if e := a.db.AddOrganizationManagedProcesses(integratorAddr, -1); e != nil {
                                log.Warnw("could not roll back managed processes counter", "error", e)
                        }
                }()
        }

        orgSigner, err := account.OrganizationSigner(a.secret, org.Creator, org.Nonce)
        if err != nil {
                errors.ErrGenericInternalServerError.Withf("could not restore organization signer: %v", err).Write(w)
                return
        }

        cspPubKey, err := a.csp.PubKey()
        if err != nil {
                errors.ErrGenericInternalServerError.Withf("could not get csp public key: %v", err).Write(w)
                return
        }

        // build + store the election metadata content-addressed; its public URL is the
        // on-chain metadata pointer (must be known before the tx, so it cannot contain
        // the not-yet-known process id).
        metaBytes, err := account.BuildElectionMetadata(draft.ElectionParams)
        if err != nil {
                errors.ErrMalformedBody.Withf("invalid election params: %v", err).Write(w)
                return
        }
        objectName, err := a.objectStorage.PutJSON(metaBytes, user.Email)
        if err != nil {
                errors.ErrInternalStorageError.WithErr(err).Write(w)
                return
        }
        // build the reference via the storage client so the URL format (and trailing-slash
        // handling) matches what the read path's local-reference match expects.
        metadataURL := a.objectStorage.LocalURL(objectName)
        // persist the reference so the read path resolves it locally without a chain round-trip.
        draft.MetadataURL = metadataURL

        // serialize build->sign->submit per organization so a concurrent publish or status
        // change for the same org cannot read the same account nonce and sign a conflicting
        // tx. The worker releases the lock after submit (held across the async hand-off);
        // every synchronous failure below releases it via the deferred unlock.
        orgLock := a.orgTxLocks.lock(org.Address)
        lockHeld := true
        defer func() {
                if lockHeld {
                        orgLock.Unlock()
                }
        }()

        // build the NewProcess tx (CSP census)
        tx, err := a.account.BuildNewProcessTx(&account.NewProcessParams{
                OrgAddress:  draft.OrgAddress,
                Params:      draft.ElectionParams,
                CensusRoot:  cspPubKey,
                CensusURI:   a.serverURL,
                MetadataURL: metadataURL,
        })
        if err != nil {
                errors.ErrMalformedBody.Withf("could not build election: %v", err).Write(w)
                return
        }

        // fund
        fundedTx, txType, err := a.account.FundTransaction(tx, orgSigner.Address())
        if err != nil {
                if apiErr, ok := err.(errors.Error); ok {
                        apiErr.Write(w)
                        return
                }
                errors.ErrVochainRequestFailed.WithErr(err).Write(w)
                return
        }
        if txType == nil || *txType != models.TxType_NEW_PROCESS {
                errors.ErrInvalidTxFormat.With("unexpected tx type for publish").Write(w)
                return
        }

        // quota / permission (same engine as the /transactions path)
        if hasPermission, err := a.subscriptions.HasTxPermission(fundedTx, *txType, org, user); !hasPermission || err != nil {
                errors.ErrUnauthorized.Withf("user does not have permission to publish: %v", err).Write(w)
                return
        }

        // sign with the organization signer
        stx, err := a.account.SignTransaction(fundedTx, orgSigner)
        if err != nil {
                errors.ErrGenericInternalServerError.Withf("could not sign election tx: %v", err).Write(w)
                return
        }

        jobID, err := apicommon.NewJobID()
        if err != nil {
                errors.ErrGenericInternalServerError.WithErr(err).Write(w)
                return
        }
        if err := a.db.CreateTxJob(jobID, db.JobTypePublishProcess, org.Address); err != nil {
                errors.ErrGenericInternalServerError.WithErr(err).Write(w)
                return
        }

        // submit + confirm on the worker pool. On success the draft becomes READY with its
        // on-chain id; on failure the PUBLISHING claim is released and any managed reservation
        // is rolled back. The idempotency guard keys off draft.Address, so a retry after
        // failure cannot create a duplicate election.
        reserved := managedReserved
        if !a.enqueueTx(txTask{jobID: jobID, run: func() (*db.JobResult, error) {
                defer orgLock.Unlock()
                data, err := a.account.SubmitSignedTx(stx)
                if err != nil {
                        if e := a.db.ClearProcessPublishing(parsedID); e != nil {
                                log.Warnw("could not clear publishing state after failed publish", "error", e)
                        }
                        if reserved {
                                if e := a.db.AddOrganizationManagedProcesses(integratorAddr, -1); e != nil {
                                        log.Warnw("could not roll back managed processes counter", "error", e)
                                }
                        }
                        return nil, err
                }
                draft.Address = internal.HexBytes(data)
                draft.Status = "READY"
                draft.PublishedAt = time.Now()
                if _, err := a.db.SetProcess(draft); err != nil {
                        return nil, err
                }
                // best-effort per-org Processes counter; only count non-test-sized elections.
                if nonTestSized {
                        if err := a.db.IncrementOrganizationProcessesCounter(org.Address); err != nil {
                                log.Warnw("could not update organization process counter", "error", err)
                        }
                }
                return &db.JobResult{Address: draft.Address, Status: "READY"}, nil
        }}) {
                // full queue: mark the job failed so it is not orphaned pending; the deferred
                // unlock, publishing-claim release and reservation rollback all fire on return.
                if e := a.db.SetJobStatus(jobID, db.JobStatusFailed, nil, "tx queue full"); e != nil {
                        log.Warnw("could not mark job failed after full queue", "error", e)
                }
                errors.ErrTxQueueFull.Write(w)
                return
        }
        // handed to a worker: it now owns the publish claim, reservation and org lock.
        committed = true
        managedReserved = false
        lockHeld = false

        apicommon.HTTPWriteJSONStatus(w, http.StatusAccepted, &apicommon.EnqueuedResponse{JobID: jobID})
}

vocdoni / saas-backend / 28449634629

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous