fix: implement email channel, satellite ingestion, credit ID validation, monotonic timestamps

project-portal/project-portal-backend/go.sum

@@ -32,6 +32,8 @@ github.com/aws/aws-sdk-go-v2/service/s3 v1.96.0 h1:oeu8VPlOre74lBA/PMhxa5vewaMIM
 github.com/aws/aws-sdk-go-v2/service/s3 v1.96.0/go.mod h1:5jggDlZ2CLQhwJBiZJb4vfk4f0GxWdEDruWKEJ1xOdo=
 github.com/aws/aws-sdk-go-v2/service/signin v1.0.5 h1:VrhDvQib/i0lxvr3zqlUwLwJP4fpmpyD9wYG1vfSu+Y=
 github.com/aws/aws-sdk-go-v2/service/signin v1.0.5/go.mod h1:k029+U8SY30/3/ras4G/Fnv/b88N4mAfliNn08Dem4M=
+github.com/aws/aws-sdk-go-v2/service/sns v1.31.3 h1:eSTEdxkfle2G98FE+Xl3db/XAXXVTJPNQo9K/Ar8oAI=
+github.com/aws/aws-sdk-go-v2/service/sns v1.31.3/go.mod h1:1dn0delSO3J69THuty5iwP0US2Glt0mx2qBBlI13pvw=
 github.com/aws/aws-sdk-go-v2/service/sso v1.30.10 h1:+VTRawC4iVY58pS/lzpo0lnoa/SYNGF4/B/3/U5ro8Y=
 github.com/aws/aws-sdk-go-v2/service/sso v1.30.10/go.mod h1:yifAsgBxgJWn3ggx70A3urX2AN49Y5sJTD1UQFlfqBw=
 github.com/aws/aws-sdk-go-v2/service/ssooidc v1.35.14 h1:0jbJeuEHlwKJ9PfXtpSFc4MF+WIWORdhN1n30ITZGFM=
@@ -306,8 +308,6 @@ gorm.io/driver/sqlite v1.6.0 h1:WHRRrIiulaPiPFmDcod6prc4l2VGVWHz80KspNsxSfQ=
 gorm.io/driver/sqlite v1.6.0/go.mod h1:AO9V1qIQddBESngQUKWL9yoH93HIeA1X6V633rBwyT8=
 gorm.io/driver/sqlserver v1.6.0 h1:VZOBQVsVhkHU/NzNhRJKoANt5pZGQAS1Bwc6m6dgfnc=
 gorm.io/driver/sqlserver v1.6.0/go.mod h1:WQzt4IJo/WHKnckU9jXBLMJIVNMVeTu25dnOzehntWw=
+gorm.io/gorm v1.25.7/go.mod h1:hbnx/Oo0ChWMn1BIhpy1oYozzpM15i4YPuHDmfYtwg8=
 gorm.io/gorm v1.31.1 h1:7CA8FTFz/gRfgqgpeKIBcervUn3xSyPUmr6B2WXJ7kg=
 gorm.io/gorm v1.31.1/go.mod h1:XyQVbO2k6YkOis7C2437jSit3SsDK72s7n7rsSHd+Gs=
-github.com/aws/aws-sdk-go-v2/service/sns v1.31.3 h1:eSTEdxkfle2G98FE+Xl3db/XAXXVTJPNQo9K/Ar8oAI=
-github.com/aws/aws-sdk-go-v2/service/sns v1.31.3/go.mod h1:1dn0delSO3J69THuty5iwP0US2Glt0mx2qBBlI13pvw=
-

project-portal/project-portal-backend/internal/monitoring/handler.go

@@ -1,7 +1,69 @@
-//go:build future
-// +build future
-
 package monitoring
 
-// This file won't be compiled in normal builds
-// Implementation pending
+import (
+	"net/http"
+	"strconv"
+
+	"github.com/gin-gonic/gin"
+)
+
+// Handler exposes monitoring endpoints on a Gin router group.
+type Handler struct {
+	svc *Service
+}
+
+// NewHandler constructs a monitoring Handler.
+func NewHandler(svc *Service) *Handler {
+	return &Handler{svc: svc}
+}
+
+// RegisterRoutes mounts the monitoring routes under the provided router group.
+//
+//	POST /api/v1/monitoring/satellite  – ingest satellite data
+//	GET  /api/v1/monitoring/:projectID – list readings for a project
+func (h *Handler) RegisterRoutes(rg *gin.RouterGroup) {
+	rg.POST("/satellite", h.ingestSatellite)
+	rg.GET("/:projectID", h.listReadings)
+}
+
+// ingestSatellite handles POST /api/v1/monitoring/satellite
+//
+//	@Summary     Ingest satellite data
+//	@Description Submit a satellite observation for a registered project
+//	@Tags        monitoring
+//	@Accept      json
+//	@Produce     json
+//	@Param       body body IngestSatelliteRequest true "Satellite reading payload"
+//	@Success     201 {object} SatelliteReading
+//	@Failure     400 {object} map[string]string
+//	@Failure     500 {object} map[string]string
+//	@Router      /api/v1/monitoring/satellite [post]
+func (h *Handler) ingestSatellite(c *gin.Context) {
+	var req IngestSatelliteRequest
+	if err := c.ShouldBindJSON(&req); err != nil {
+		c.JSON(http.StatusBadRequest, gin.H{"error": err.Error()})
+		return
+	}
+
+	reading, err := h.svc.IngestSatellite(c.Request.Context(), req)
+	if err != nil {
+		c.JSON(http.StatusBadRequest, gin.H{"error": err.Error()})
+		return
+	}
+
+	c.JSON(http.StatusCreated, reading)
+}
+
+// listReadings handles GET /api/v1/monitoring/:projectID
+func (h *Handler) listReadings(c *gin.Context) {
+	projectID := c.Param("projectID")
+	limit, _ := strconv.Atoi(c.DefaultQuery("limit", "50"))
+
+	readings, err := h.svc.ListReadings(c.Request.Context(), projectID, limit)
+	if err != nil {
+		c.JSON(http.StatusInternalServerError, gin.H{"error": err.Error()})
+		return
+	}
+
+	c.JSON(http.StatusOK, gin.H{"data": readings, "count": len(readings)})
+}

project-portal/project-portal-backend/internal/monitoring/ingestion/satellite.go

@@ -1,7 +1,155 @@
-//go:build future
-// +build future
+package ingestion
 
-package monitoring
+import (
+	"context"
+	"errors"
+	"fmt"
+	"strings"
+	"time"
 
-// This file won't be compiled in normal builds
-// Implementation pending
+	"github.com/google/uuid"
+)
+
+// SatelliteReading is a local copy of the reading type to avoid import cycles.
+// The monitoring package embeds this via its own SatelliteReading type.
+type SatelliteReading struct {
+	ID          string
+	ProjectID   string
+	Source      string
+	DataType    string
+	NDVIMean    *float64
+	NDVIMin     *float64
+	NDVIMax     *float64
+	BiomassTons *float64
+	ImageryURL  string
+	BoundingBox *BoundingBox
+	Metadata    map[string]string
+	CapturedAt  time.Time
+	IngestedAt  time.Time
+}
+
+// BoundingBox is a geographic bounding box (WGS-84).
+type BoundingBox struct {
+	MinLat float64 `json:"min_lat"`
+	MaxLat float64 `json:"max_lat"`
+	MinLon float64 `json:"min_lon"`
+	MaxLon float64 `json:"max_lon"`
+}
+
+// IngestRequest is the validated input for the satellite pipeline.
+type IngestRequest struct {
+	ProjectID   string
+	Source      string
+	DataType    string
+	NDVIMean    *float64
+	NDVIMin     *float64
+	NDVIMax     *float64
+	BiomassTons *float64
+	ImageryURL  string
+	BoundingBox *BoundingBox
+	Metadata    map[string]string
+	CapturedAt  time.Time
+}
+
+// Repository is the persistence contract used by the pipeline.
+type Repository interface {
+	Save(ctx context.Context, r *SatelliteReading) error
+}
+
+// allowedSources lists accepted satellite data providers.
+var allowedSources = map[string]bool{
+	"sentinel-2":  true,
+	"planet-labs": true,
+	"landsat-8":   true,
+	"landsat-9":   true,
+	"drone":       true,
+}
+
+// allowedDataTypes lists accepted data type identifiers.
+var allowedDataTypes = map[string]bool{
+	"NDVI":    true,
+	"BIOMASS": true,
+	"IMAGERY": true,
+}
+
+// SatellitePipeline validates and persists incoming satellite data.
+type SatellitePipeline struct {
+	repo Repository
+}
+
+// NewSatellitePipeline constructs a SatellitePipeline backed by the given repository.
+func NewSatellitePipeline(repo Repository) *SatellitePipeline {
+	return &SatellitePipeline{repo: repo}
+}
+
+// Ingest validates the request and persists a SatelliteReading.
+func (p *SatellitePipeline) Ingest(ctx context.Context, req IngestRequest) (*SatelliteReading, error) {
+	if err := validate(req); err != nil {
+		return nil, err
+	}
+
+	now := time.Now().UTC()
+	reading := &SatelliteReading{
+		ID:          uuid.NewString(),
+		ProjectID:   req.ProjectID,
+		Source:      strings.ToLower(req.Source),
+		DataType:    strings.ToUpper(req.DataType),
+		NDVIMean:    req.NDVIMean,
+		NDVIMin:     req.NDVIMin,
+		NDVIMax:     req.NDVIMax,
+		BiomassTons: req.BiomassTons,
+		ImageryURL:  req.ImageryURL,
+		BoundingBox: req.BoundingBox,
+		Metadata:    req.Metadata,
+		CapturedAt:  req.CapturedAt.UTC(),
+		IngestedAt:  now,
+	}
+
+	if err := p.repo.Save(ctx, reading); err != nil {
+		return nil, fmt.Errorf("satellite ingestion: persist failed: %w", err)
+	}
+
+	return reading, nil
+}
+
+// validate checks required fields, allowed values, and data-type-specific constraints.
+func validate(req IngestRequest) error {
+	if strings.TrimSpace(req.ProjectID) == "" {
+		return errors.New("project_id is required")
+	}
+	if !allowedSources[strings.ToLower(req.Source)] {
+		return fmt.Errorf("unsupported source %q; allowed: sentinel-2, planet-labs, landsat-8, landsat-9, drone", req.Source)
+	}
+	if !allowedDataTypes[strings.ToUpper(req.DataType)] {
+		return fmt.Errorf("unsupported data_type %q; allowed: NDVI, BIOMASS, IMAGERY", req.DataType)
+	}
+	if req.CapturedAt.IsZero() {
+		return errors.New("captured_at is required")
+	}
+	if req.CapturedAt.After(time.Now().UTC().Add(5 * time.Minute)) {
+		return errors.New("captured_at cannot be in the future")
+	}
+
+	switch strings.ToUpper(req.DataType) {
+	case "NDVI":
+		if req.NDVIMean == nil {
+			return errors.New("ndvi_mean is required for NDVI data type")
+		}
+		if *req.NDVIMean < -1 || *req.NDVIMean > 1 {
+			return errors.New("ndvi_mean must be in range [-1, 1]")
+		}
+	case "BIOMASS":
+		if req.BiomassTons == nil {
+			return errors.New("biomass_tons is required for BIOMASS data type")
+		}
+		if *req.BiomassTons < 0 {
+			return errors.New("biomass_tons must be non-negative")
+		}
+	case "IMAGERY":
+		if strings.TrimSpace(req.ImageryURL) == "" {
+			return errors.New("imagery_url is required for IMAGERY data type")
+		}
+	}
+
+	return nil
+}

project-portal/project-portal-backend/internal/monitoring/models.go

@@ -1,7 +1,26 @@
-//go:build future
-// +build future
-
 package monitoring
 
-// This file won't be compiled in normal builds
-// Implementation pending
+import (
+	"time"
+
+	"carbon-scribe/project-portal/project-portal-backend/internal/monitoring/ingestion"
+)
+
+// Re-export ingestion types so callers only import this package.
+type BoundingBox = ingestion.BoundingBox
+type SatelliteReading = ingestion.SatelliteReading
+
+// IngestSatelliteRequest is the API payload for POST /api/v1/monitoring/satellite.
+type IngestSatelliteRequest struct {
+	ProjectID   string            `json:"project_id" binding:"required"`
+	Source      string            `json:"source" binding:"required"`
+	DataType    string            `json:"data_type" binding:"required"`
+	NDVIMean    *float64          `json:"ndvi_mean"`
+	NDVIMin     *float64          `json:"ndvi_min"`
+	NDVIMax     *float64          `json:"ndvi_max"`
+	BiomassTons *float64          `json:"biomass_tons"`
+	ImageryURL  string            `json:"imagery_url"`
+	BoundingBox *BoundingBox      `json:"bounding_box"`
+	Metadata    map[string]string `json:"metadata"`
+	CapturedAt  time.Time         `json:"captured_at" binding:"required"`
+}

project-portal/project-portal-backend/internal/monitoring/postgres_repository.go

@@ -1,7 +1,86 @@
-//go:build future
-// +build future
-
 package monitoring
 
-// This file won't be compiled in normal builds
-// Implementation pending
+import (
+	"context"
+	"database/sql"
+	"encoding/json"
+	"fmt"
+
+	"carbon-scribe/project-portal/project-portal-backend/internal/monitoring/ingestion"
+)
+
+// PostgresRepository implements Repository using a *sql.DB connection.
+type PostgresRepository struct {
+	db *sql.DB
+}
+
+// NewPostgresRepository constructs a PostgresRepository.
+func NewPostgresRepository(db *sql.DB) *PostgresRepository {
+	return &PostgresRepository{db: db}
+}
+
+// Save inserts a SatelliteReading into the satellite_readings table.
+func (r *PostgresRepository) Save(ctx context.Context, reading *ingestion.SatelliteReading) error {
+	metaJSON, err := json.Marshal(reading.Metadata)
+	if err != nil {
+		return fmt.Errorf("marshal metadata: %w", err)
+	}
+
+	var bboxJSON []byte
+	if reading.BoundingBox != nil {
+		bboxJSON, err = json.Marshal(reading.BoundingBox)
+		if err != nil {
+			return fmt.Errorf("marshal bounding_box: %w", err)
+		}
+	}
+
+	_, err = r.db.ExecContext(ctx, `
+		INSERT INTO satellite_readings
+			(id, project_id, source, data_type, ndvi_mean, ndvi_min, ndvi_max,
+			 biomass_tons, imagery_url, bounding_box, metadata, captured_at, ingested_at)
+		VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12,$13)`,
+		reading.ID, reading.ProjectID, reading.Source, reading.DataType,
+		reading.NDVIMean, reading.NDVIMin, reading.NDVIMax,
+		reading.BiomassTons, reading.ImageryURL, bboxJSON, metaJSON,
+		reading.CapturedAt, reading.IngestedAt,
+	)
+	return err
+}
+
+// ListByProject returns the most recent satellite readings for a project.
+func (r *PostgresRepository) ListByProject(ctx context.Context, projectID string, limit int) ([]ingestion.SatelliteReading, error) {
+	rows, err := r.db.QueryContext(ctx, `
+		SELECT id, project_id, source, data_type, ndvi_mean, ndvi_min, ndvi_max,
+		       biomass_tons, imagery_url, bounding_box, metadata, captured_at, ingested_at
+		FROM satellite_readings
+		WHERE project_id = $1
+		ORDER BY captured_at DESC
+		LIMIT $2`, projectID, limit)
+	if err != nil {
+		return nil, err
+	}
+	defer rows.Close()
+
+	var results []ingestion.SatelliteReading
+	for rows.Next() {
+		var sr ingestion.SatelliteReading
+		var bboxJSON, metaJSON []byte
+		if err := rows.Scan(
+			&sr.ID, &sr.ProjectID, &sr.Source, &sr.DataType,
+			&sr.NDVIMean, &sr.NDVIMin, &sr.NDVIMax,
+			&sr.BiomassTons, &sr.ImageryURL, &bboxJSON, &metaJSON,
+			&sr.CapturedAt, &sr.IngestedAt,
+		); err != nil {
+			return nil, err
+		}
+		if len(bboxJSON) > 0 {
+			sr.BoundingBox = &ingestion.BoundingBox{}
+			_ = json.Unmarshal(bboxJSON, sr.BoundingBox)
+		}
+		if len(metaJSON) > 0 {
+			_ = json.Unmarshal(metaJSON, &sr.Metadata)
+		}
+		results = append(results, sr)
+	}
+	return results, rows.Err()
+}

project-portal/project-portal-backend/internal/monitoring/repository.go

@@ -1,7 +1,13 @@
-//go:build future
-// +build future
-
 package monitoring
 
-// This file won't be compiled in normal builds
-// Implementation pending
+import (
+	"context"
+
+	"carbon-scribe/project-portal/project-portal-backend/internal/monitoring/ingestion"
+)
+
+// Repository defines persistence operations for satellite monitoring data.
+type Repository interface {
+	ingestion.Repository
+	ListByProject(ctx context.Context, projectID string, limit int) ([]SatelliteReading, error)
+}