28188283899

Committed 25 Jun 2026 05:24PM UTC coverage: 62.203% (+0.1%) from 62.083%

Build # 28188283899

Build Type

push

github

Committed by

web-flow

Commit Message

fix(api): resolve election metadata from a stored reference on GET /process/{id} (#553)

* fix(api): resolve election metadata from a stored reference on GET /process/{id}

The Vochain node only resolves ipfs:// metadata pointers, so the https
/storage pointer this service publishes is never resolved downstream and the
election metadata goes missing.

Store a generic reference (Process.MetadataURL) and resolve it on read:
- our own object storage (matched first, works even when serverURL is empty) ->
  read locally;
- ipfs:// -> resolve via the Vochain, then cache locally and promote the
  reference so later reads stay local;
- other http(s) -> best-effort external fetch (3s timeout, 1 MiB cap).

The reference is bootstrapped from the on-chain pointer when missing (covers
pre-existing/external processes, no migration). Resolution is best-effort: any
error logs and returns the stored value. Drafts (no on-chain address) keep
their stored metadata map.

* fix(api): cache externally-fetched metadata too, not just ipfs

Metadata is immutable (a change is a republish through this API), so the
external http(s) case is as safe to cache as ipfs. Unify both remote branches:
any remotely-resolved document is stored locally and the reference promoted, so
later reads resolve from our own storage with no external/Vochain round-trip.

* fix(api): address Copilot review on metadata resolution

- match relative /storage/ references and trim a trailing slash from serverURL
  so local resolution survives a serverURL change/misconfig. This lives in
  objectstorage (Client.LocalName / Client.LocalURL) so URL parse+format stay
  with the storage layer, not the api package.
- persist the promoted reference with a targeted SetProcessMetadataURL ($set
  metadataURL only) instead of a full SetProcess rewrite, so the read path can't
  clobber concurrent updates to other process fields.
- add tests: fetchExternalMetadata (ok/non-200/over-cap) and the remote->local
  cache+promot... (continued)

Coverage Stats

91 of 119 new or added lines in 4 files covered. (76.47%)

9909 of 15930 relevant lines covered (62.2%)

44.47 hits per line

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

53.03

/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                                "24-hex draft ProcessID"
//        @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                409                {object}        errors.Error                                        "Process already exists"
//        @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.
//        @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
        }

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

// 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 voting process
//        @Description        Delete a voting process. Requires Manager/Admin role.
//        @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                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
        }

        // 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)
}

// 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 per-org and aggregate election/census quotas.
//        @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/census 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.
        managedReserved := false
        var integratorAddr common.Address
        nonTestSized := draft.ElectionParams.MaxCensusSize > uint64(db.TestMaxCensusSize)
        if org.ManagedBy != (common.Address{}) && nonTestSized {
                integrator, err := a.db.Organization(org.ManagedBy)
                if err != nil {
                        errors.ErrGenericInternalServerError.Withf("could not get integrator organization: %v", err).Write(w)
                        return
                }
                limits, err := a.subscriptions.EffectiveIntegratorLimits(integrator)
                if err != nil {
                        if apiErr, ok := err.(errors.Error); ok {
                                apiErr.Write(w)
                                return
                        }
                        errors.ErrGenericInternalServerError.WithErr(err).Write(w)
                        return
                }
                if err := a.db.ReserveManagedPublish(integrator.Address,
                        limits.MaxManagedProcesses, limits.MaxManagedCensusSize, int(draft.ElectionParams.MaxCensusSize)); err != nil {
                        if err == db.ErrManagedQuotaReached {
                                errors.ErrIntegratorQuotaExceeded.Write(w)
                                return
                        }
                        errors.ErrGenericInternalServerError.WithErr(err).Write(w)
                        return
                }
                managedReserved = true
                integratorAddr = integrator.Address
                // 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)
                        }
                        if e := a.db.AddOrganizationManagedCensusSize(integratorAddr,
                                -int64(draft.ElectionParams.MaxCensusSize)); e != nil {
                                log.Warnw("could not roll back managed census 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
        censusSize := int64(draft.ElectionParams.MaxCensusSize)
        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)
                                }
                                if e := a.db.AddOrganizationManagedCensusSize(integratorAddr, -censusSize); e != nil {
                                        log.Warnw("could not roll back managed census 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 / 28188283899

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