Azure Databricks
Platform Engineering

trigger:
  branches:
    include: [main]
  paths:
    include: [src/**, Dockerfile]

variables:
  acrName:     crneondeck
  imageName:   casino-app
  aksCluster:  aks-neondeck-prod
  canaryPct:   5

stages:
# ── Build: Docker image with layer caching ──
- stage: Build
  jobs:
  - job: DockerBuild
    pool: { vmImage: ubuntu-latest }
    steps:
    - task: Docker@2
      inputs:
        containerRegistry: $(acrName)
        repository:        $(imageName)
        command:           buildAndPush
        Dockerfile:        Dockerfile
        tags:              $(Build.BuildId)
        arguments:         --cache-from $(acrName).azurecr.io/$(imageName):latest

# ── SAST: Semgrep security scan with fail threshold ──
- stage: SAST
  dependsOn: Build
  jobs:
  - job: SemgrepScan
    steps:
    - script: |
        pip install semgrep
        semgrep --config=p/owasp-top-ten --config=p/typescript \
          --error --severity ERROR \
          --json --output semgrep-results.json \
          src/
      displayName: Semgrep OWASP Scan
    - task: PublishBuildArtifacts@1
      inputs: { PathtoPublish: semgrep-results.json, ArtifactName: sast-report }
      condition: always()

# ── Deploy: Canary rollout with health check ──
- stage: CanaryDeploy
  dependsOn: SAST
  jobs:
  - deployment: Canary
    environment: production
    strategy:
      canary:
        increments: [5, 25, 50, 100]
        deploy:
          steps:
          - script: |
              kubectl set image deployment/casino-app \
                casino-app=$(acrName).azurecr.io/$(imageName):$(Build.BuildId)
              kubectl rollout status deployment/casino-app --timeout=300s
        routeTraffic:
          steps:
          - script: |
              kubectl annotate ingress casino-app \
                nginx.ingress.kubernetes.io/canary-weight="$(strategy.increment)" --overwrite
        postRouteTraffic:
          steps:
          - script: |  # Health check: 15-min error-rate window
              BASELINE_ERR=$(curl -s 'http://prometheus:9090/api/v1/query?query=rate(http_errors[15m])' | jq '.data.result[0].value[1]')
              CANARY_ERR=$(curl -s 'http://prometheus:9090/api/v1/query?query=rate(http_errors{canary="true"}[15m])' | jq '.data.result[0].value[1]')
              if (( $(echo "$CANARY_ERR > $BASELINE_ERR * 2" | bc -l) )); then
                echo "##vso[task.logissue type=error]Canary error rate 2x baseline — rolling back"
                exit 1
              fi
        on:
          failure:
            steps:
            - script: kubectl rollout undo deployment/casino-app
              displayName: Rollback canary

import mlflow
import mlflow.sklearn
from pyspark.sql import SparkSession
from sklearn.ensemble import GradientBoostingClassifier
from sklearn.metrics import precision_score, recall_score, f1_score
from mlflow.tracking import MlflowClient

spark = SparkSession.builder.appName("wab-credit-risk").getOrCreate()
client = MlflowClient()

# ── 1. Load data from Delta Lake feature store ──
df = spark.read.format("delta").table("wab_catalog.gold.credit_risk_features")
feature_cols = ["transaction_amount", "acct_days_since_open", "avg_daily_balance",
                "overdraft_count_30d", "credit_utilization_pct"]
pdf = df.select(feature_cols + ["is_default"]).toPandas()

X = pdf[feature_cols]
y = pdf["is_default"]

# ── 2. Train with MLflow tracking ──
mlflow.set_experiment("/wab/credit-risk-scorecard")
with mlflow.start_run(run_name="gbm-v2.4") as run:
    model = GradientBoostingClassifier(
        n_estimators=500, max_depth=6, learning_rate=0.05,
        subsample=0.8, min_samples_leaf=20
    )
    model.fit(X, y)
    preds = model.predict(X)

    # Log metrics
    mlflow.log_metric("precision", precision_score(y, preds))
    mlflow.log_metric("recall",    recall_score(y, preds))
    mlflow.log_metric("f1",        f1_score(y, preds))
    mlflow.log_param("n_estimators", 500)
    mlflow.sklearn.log_model(model, "credit-risk-model")

# ── 3. Model registry: promote staging → production ──
model_name = "wab-credit-risk-scorecard"
model_uri  = f"runs:/{run.info.run_id}/credit-risk-model"
mv = mlflow.register_model(model_uri, model_name)

# Compare against current production
prod_versions = client.get_latest_versions(model_name, stages=["Production"])
if prod_versions:
    prod_run = client.get_run(prod_versions[0].run_id)
    prod_f1  = prod_run.data.metrics["f1"]
    new_f1   = f1_score(y, preds)
    if new_f1 > prod_f1:
        client.transition_model_version_stage(
            model_name, mv.version, stage="Production",
            archive_existing_versions=True
        )
        print(f"Promoted v{mv.version}: F1 {new_f1:.4f} > {prod_f1:.4f}")
    else:
        client.transition_model_version_stage(
            model_name, mv.version, stage="Staging"
        )
        print(f"Kept in staging: F1 {new_f1:.4f} ≤ {prod_f1:.4f}")
else:
    client.transition_model_version_stage(
        model_name, mv.version, stage="Production"
    )
    print(f"First production model: v{mv.version}")

# SCIM 2.0 User provisioning endpoint for Entra ID → internal systems
from flask import Flask, request, jsonify
import uuid
from datetime import datetime

app = Flask(__name__)

SCIM_SCHEMA = "urn:ietf:params:scim:schemas:core:2.0:User"

def map_scim_to_internal(scim_user: dict) -> dict:
    """Map SCIM user attributes to internal schema."""
    name = scim_user.get("name", {})
    emails = scim_user.get("emails", [{}])
    primary_email = next(
        (e["value"] for e in emails if e.get("primary")),
        emails[0].get("value", "") if emails else ""
    )
    return {
        "external_id": scim_user.get("externalId"),
        "username":    scim_user.get("userName"),
        "email":       primary_email,
        "first_name":  name.get("givenName", ""),
        "last_name":   name.get("familyName", ""),
        "active":      scim_user.get("active", True),
    }


@app.route("/scim/v2/Users", methods=["POST"])
def create_user():
    scim_user = request.get_json()

    # Validate SCIM schema
    schemas = scim_user.get("schemas", [])
    if SCIM_SCHEMA not in schemas:
        return jsonify({
            "schemas": ["urn:ietf:params:scim:api:messages:2.0:Error"],
            "detail":  "Missing required schema",
            "status":  400
        }), 400

    internal = map_scim_to_internal(scim_user)

    # Conflict detection: check if user already exists
    existing = db.users.find_one({"email": internal["email"]})
    if existing:
        return jsonify({
            "schemas": ["urn:ietf:params:scim:api:messages:2.0:Error"],
            "detail":  f"User {internal['email']} already exists",
            "status":  409
        }), 409

    # Provision in downstream systems
    user_id = str(uuid.uuid4())
    internal["id"] = user_id
    internal["created_at"] = datetime.utcnow().isoformat() + "Z"
    db.users.insert_one(internal)

    # SCIM-compliant response
    return jsonify({
        "schemas":    [SCIM_SCHEMA],
        "id":         user_id,
        "externalId": internal["external_id"],
        "userName":   internal["username"],
        "name":       {"givenName": internal["first_name"],
                       "familyName": internal["last_name"]},
        "emails":     [{"value": internal["email"], "primary": True}],
        "active":     internal["active"],
        "meta": {
            "resourceType": "User",
            "created":      internal["created_at"],
            "location":     f"/scim/v2/Users/{user_id}"
        }
    }), 201