Skip to content
Nightly Dataset Sync (production -> dev)

Nightly Dataset Sync (production -> dev) #16

Sign in to view logs

Nightly Dataset Sync (production -> dev)

Nightly Dataset Sync (production -> dev) #16

Workflow file for this run

.github/workflows/nightly-dataset-sync.yml at 35ca735
name: Nightly Dataset Sync (production -> dev)
on:
schedule:
- cron: "0 4 * * *" # Every day at 04:00 UTC
workflow_dispatch:
jobs:
sync-dataset:
runs-on: ubuntu-latest
environment: production
steps:
- uses: actions/checkout@v4
- uses: pnpm/action-setup@v4
- uses: actions/setup-node@v4
with:
node-version-file: ".nvmrc"
cache: "pnpm"
- name: Install dependencies
run: pnpm install --frozen-lockfile
- name: Sync production dataset to dev
env:
SANITY_STUDIO_PROJECT_ID: ${{ vars.SANITY_STUDIO_PROJECT_ID || 'hfh83o0w' }}
SANITY_AUTH_TOKEN: ${{ secrets.SANITY_AUTH_TOKEN }}
SANITY_API_WRITE_TOKEN: ${{ secrets.SANITY_API_WRITE_TOKEN }}
run: |
PROJECT_ID="${SANITY_STUDIO_PROJECT_ID:-hfh83o0w}"
API_HOST="https://api.sanity.io"
API_VERSION="v2021-06-07"
echo "1. Ensuring 'dev' dataset exists (public ACL)..."
for T in "$SANITY_AUTH_TOKEN" "$SANITY_API_WRITE_TOKEN"; do
[ -z "$T" ] && continue
RESP=$(curl -s -X PUT "$API_HOST/$API_VERSION/projects/$PROJECT_ID/datasets/dev" \
-H "Authorization: Bearer $T" \
-H "Content-Type: application/json" \
-d '{"aclMode":"public"}' || true)
echo "Dataset create response: $RESP"
if echo "$RESP" | grep -q '"datasetName":"dev"'; then
break
fi
done
echo "2. Exporting 'production' dataset to /tmp/production.ndjson..."
curl -f -sS "https://$PROJECT_ID.api.sanity.io/$API_VERSION/data/export/production" -o /tmp/production.ndjson
gzip -k -f /tmp/production.ndjson
wc -l /tmp/production.ndjson
echo "3. Importing documents and referenced image binaries into 'dev' dataset..."
cat << 'EOF' > /tmp/sync.cjs
const fs = require("fs");
const projectId = process.env.SANITY_STUDIO_PROJECT_ID || "hfh83o0w";
const tokens = [process.env.SANITY_API_WRITE_TOKEN, process.env.SANITY_AUTH_TOKEN].filter(Boolean);
const lines = fs.readFileSync("/tmp/production.ndjson", "utf8").trim().split("\n").filter(Boolean);
const allDocs = lines.map((l) => JSON.parse(l));
const assetDocs = allDocs.filter((d) => d._type === "sanity.imageAsset" || d._type === "sanity.fileAsset");
const imageAssetMap = new Map(
allDocs.filter((d) => d._type === "sanity.imageAsset").map((d) => [d._id, d])
);
const contentDocs = allDocs.filter((d) => d._type !== "sanity.imageAsset" && d._type !== "sanity.fileAsset");
// Collect all known IDs (including assets and syndication.* targets) so valid strong references are never weakened
const allTargetIds = new Set(allDocs.map((d) => d._id));
const syndicationStubs = new Map();
for (const doc of contentDocs) {
if (Array.isArray(doc.syndications)) {
for (const item of doc.syndications) {
if (item && typeof item._ref === "string" && item._ref.startsWith("syndication.")) {
allTargetIds.add(item._ref);
if (!contentDocs.some((d) => d._id === item._ref)) {
const platform = item._ref.endsWith(".devto") ? "devto" : "hashnode";
const parentId = doc._id.replace(/^drafts\./, "");
syndicationStubs.set(item._ref, {
_id: item._ref,
_type: "syndication",
platform,
status: "simulated_complete",
externalUrl: `https://${platform === "devto" ? "dev.to" : "hashnode.com"}/codingcatdev`,
syndicatedAt: doc._updatedAt || new Date().toISOString(),
target: { _type: "reference", _ref: parentId, _weak: true },
});
}
}
}
}
}
// Track any production documents that have duplicate syndications keys so we can clean them up in production too
const prodDeduplicateMutations = [];
for (const doc of contentDocs) {
if (Array.isArray(doc.syndications) && doc.syndications.length > 0) {
const seenRefs = new Set();
const seenKeys = new Set();
const deduped = [];
let changed = false;
for (const item of doc.syndications) {
if (!item || typeof item._ref !== "string") continue;
if (seenRefs.has(item._ref)) {
changed = true;
continue;
}
seenRefs.add(item._ref);
const detKey = `syn_${item._ref.replace(/[^a-zA-Z0-9]/g, "_")}`;
const key = item._key && !seenKeys.has(item._key) ? item._key : detKey;
if (key !== item._key || item._weak) {
changed = true;
}
seenKeys.add(key);
deduped.push({
_key: key,
_type: "reference",
_ref: item._ref,
});
}
if (changed) {
doc.syndications = deduped;
prodDeduplicateMutations.push({
patch: {
id: doc._id,
set: { syndications: deduped },
},
});
}
}
}
function sanitizeDoc(obj) {
if (!obj || typeof obj !== "object") return obj;
if (Array.isArray(obj)) {
const mapped = obj.map(sanitizeDoc);
const usedKeys = new Set();
const deduped = [];
for (const item of mapped) {
if (item && typeof item === "object" && typeof item._key === "string") {
if (usedKeys.has(item._key)) {
if (typeof item._ref === "string") {
// Duplicate reference key in array: skip duplicate entry
continue;
}
item._key = `${item._key}_${usedKeys.size}`;
}
usedKeys.add(item._key);
}
deduped.push(item);
}
return deduped;
}
const copy = {};
for (const [k, v] of Object.entries(obj)) {
if (k === "_rev") continue;
copy[k] = sanitizeDoc(v);
}
if (typeof copy._ref === "string") {
if (allTargetIds.has(copy._ref)) {
delete copy._weak;
delete copy._strengthenOnPublish;
} else {
copy._weak = true;
}
}
return copy;
}
const sanitized = contentDocs.map(sanitizeDoc);
console.log(`Prepared ${sanitized.length} content documents and ${syndicationStubs.size} syndication stubs for dataset "dev"...`);
let activeToken = tokens[0];
async function mutateBatch(mutations, dataset = "dev") {
for (let attempt = 0; attempt < 5; attempt++) {
const token = activeToken || tokens[attempt % tokens.length];
const res = await fetch(`https://${projectId}.api.sanity.io/v2021-06-07/data/mutate/${dataset}?visibility=deferred&autoGenerateArrayKeys=true`, {
method: "POST",
headers: {
Authorization: `Bearer ${token}`,
"Content-Type": "application/json",
},
body: JSON.stringify({ mutations }),
});
if (res.ok) {
activeToken = token;
return;
}
const errText = await res.text();
if (res.status === 401 || res.status === 403) {
activeToken = tokens.find((t) => t !== token) || token;
continue;
}
if (res.status === 429 || res.status >= 500) {
const delay = 1000 * Math.pow(2, attempt);
console.warn(`Rate limited / server busy (HTTP ${res.status}), retrying in ${delay}ms...`);
await new Promise((r) => setTimeout(r, delay));
continue;
}
throw new Error(`Mutate batch (${dataset}) failed: HTTP ${res.status}: ${errText}`);
}
throw new Error(`Mutate batch (${dataset}) failed after 5 retries.`);
}
async function sendChunked(items, label, op = "createOrReplace") {
let batch = [];
let batchBytes = 0;
let sent = 0;
const MAX_ITEMS = 200;
const MAX_BYTES = 2500000;
for (const item of items) {
const mutation = { [op]: item };
const itemBytes = Buffer.byteLength(JSON.stringify(mutation), "utf8");
if (batch.length > 0 && (batch.length >= MAX_ITEMS || batchBytes + itemBytes > MAX_BYTES)) {
await mutateBatch(batch);
sent += batch.length;
if (sent % 500 < batch.length) {
console.log(`[${label}] Imported ${sent} / ${items.length}...`);
}
batch = [];
batchBytes = 0;
}
batch.push(mutation);
batchBytes += itemBytes;
}
if (batch.length > 0) {
await mutateBatch(batch);
sent += batch.length;
}
console.log(`[${label}] Completed (${sent} / ${items.length}).`);
}
async function uploadMissingDevImages() {
const q = encodeURIComponent(`*[_type == "sanity.imageAsset" && path match "images/${projectId}/dev/*"]._id`);
let devUploadedIds = new Set();
for (const token of tokens) {
const res = await fetch(`https://${projectId}.api.sanity.io/v2025-02-19/data/query/dev?query=${q}`, {
headers: { Authorization: `Bearer ${token}` },
});
if (res.ok) {
const json = await res.json();
devUploadedIds = new Set(json.result || []);
break;
}
}
console.log(`Found ${devUploadedIds.size} image binaries already in 'dev' CDN bucket.`);
const activeTypes = new Set(["post", "podcast", "guest", "author", "sponsor", "page", "settings", "podcastType"]);
const priorityRefs = new Set();
const inlineRefs = new Set();
function collectRefs(obj) {
if (!obj || typeof obj !== "object") return;
if (Array.isArray(obj)) return obj.forEach(collectRefs);
if (typeof obj._ref === "string" && obj._ref.startsWith("image-")) {
inlineRefs.add(obj._ref);
}
for (const v of Object.values(obj)) collectRefs(v);
}
for (const doc of contentDocs) {
if (!activeTypes.has(doc._type)) continue;
if (doc.coverImage?.asset?._ref) priorityRefs.add(doc.coverImage.asset._ref);
if (doc.photo?.asset?._ref) priorityRefs.add(doc.photo.asset._ref);
if (doc.ogImage?.asset?._ref) priorityRefs.add(doc.ogImage.asset._ref);
collectRefs(doc);
}
const orderedRefs = [...new Set([...priorityRefs, ...inlineRefs])].filter(
(id) => !devUploadedIds.has(id) && imageAssetMap.has(id)
);
console.log(`Uploading ${orderedRefs.length} missing active image binaries to 'dev' dataset...`);
let uploaded = 0;
let idx = 0;
const CONCURRENCY = 20;
async function worker() {
while (idx < orderedRefs.length) {
const currentIdx = idx++;
const assetId = orderedRefs[currentIdx];
const assetDoc = imageAssetMap.get(assetId);
if (!assetDoc?.url) continue;
for (let attempt = 0; attempt < 4; attempt++) {
try {
const imgRes = await fetch(assetDoc.url);
if (!imgRes.ok) break;
const buf = Buffer.from(await imgRes.arrayBuffer());
const token = activeToken || tokens[attempt % tokens.length];
const filename = encodeURIComponent(assetDoc.originalFilename || `${assetId}.${assetDoc.extension || "jpg"}`);
const upRes = await fetch(`https://${projectId}.api.sanity.io/v2021-06-07/assets/images/dev?filename=${filename}`, {
method: "POST",
headers: {
Authorization: `Bearer ${token}`,
"Content-Type": assetDoc.mimeType || imgRes.headers.get("content-type") || "image/jpeg",
},
body: buf,
});
if (upRes.ok) {
activeToken = token;
devUploadedIds.add(assetId);
uploaded++;
if (uploaded % 100 === 0 || uploaded === orderedRefs.length) {
console.log(`[Image Binaries] Uploaded ${uploaded} / ${orderedRefs.length}...`);
}
break;
}
if (upRes.status === 401 || upRes.status === 403) {
activeToken = tokens.find((t) => t !== token) || token;
continue;
}
if (upRes.status === 429 || upRes.status >= 500) {
await new Promise((r) => setTimeout(r, 500 * Math.pow(2, attempt)));
continue;
}
break;
} catch (err) {
await new Promise((r) => setTimeout(r, 500));
}
}
}
}
await Promise.all(Array.from({ length: CONCURRENCY }, () => worker()));
console.log(`[Image Binaries] Finished uploading ${uploaded} image binaries to 'dev'.`);
return devUploadedIds;
}
(async () => {
if (prodDeduplicateMutations.length > 0) {
console.log(`Cleaning up ${prodDeduplicateMutations.length} duplicate syndications arrays in production...`);
try {
await mutateBatch(prodDeduplicateMutations, "production");
console.log("Production syndications arrays deduplicated.");
} catch (prodErr) {
console.warn("Could not patch production syndications:", prodErr.message);
}
}
// Upload missing active image binaries to dev (creates real sanity.imageAsset docs with dev bucket paths)
const devUploadedIds = await uploadMissingDevImages();
// Import remaining asset metadata docs (only those not already uploaded as dev binaries) so all asset refs resolve
const remainingAssetDocs = assetDocs
.filter((d) => !devUploadedIds.has(d._id))
.map(({ _rev, ...rest }) => rest);
if (remainingAssetDocs.length > 0) {
await sendChunked(remainingAssetDocs, "Pass 1a: Remaining Asset Metadata", "createIfNotExists");
}
// Create content + syndication stubs so all internal document references resolve as strong references
const stubs = [
...sanitized.map((d) => ({ _id: d._id, _type: d._type })),
...Array.from(syndicationStubs.values()),
];
await sendChunked(stubs, "Pass 1b: Document & Syndication Stubs", "createIfNotExists");
// Populate full content documents (with strong coverImage.asset & syndications references and unique _keys)
await sendChunked(sanitized, "Pass 2: Content Docs", "createOrReplace");
console.log("Dataset sync completed successfully!");
})().catch((err) => {
console.error(err);
process.exit(1);
});
EOF
node /tmp/sync.cjs
- name: Upload Sanity production dataset backup artifact
uses: actions/upload-artifact@v4
with:
name: sanity-production-backup-${{ github.run_id }}
path: /tmp/production.ndjson.gz
retention-days: 30