Connect MongoDB to PostPenguin

Store PostPenguin blog posts in MongoDB. Perfect for Node.js applications, serverless functions, or any stack that uses MongoDB.

🗄️ Collection Schema

MongoDB is schema-flexible, but here's a recommended document structure:

// posts collection document structure
{
  "_id": ObjectId("..."),
  "postPenguinId": "pp_abc123",        // Unique ID from PostPenguin
  "title": "Your Blog Post Title",
  "slug": "your-blog-post-title",       // URL-friendly slug
  "html": "<p className="text-gray-700">Full HTML content...</p>",
  "metaTitle": "SEO Title",
  "metaDescription": "SEO description...",
  "featuredImage": "https://...",
  "tags": ["tag1", "tag2"],
  "status": "publish",
  "publishedAt": ISODate("2025-12-02T12:00:00Z"),
  "createdAt": ISODate("2025-12-02T12:00:00Z"),
  "updatedAt": ISODate("2025-12-02T12:00:00Z")
}

Create Indexes

// Run in MongoDB shell or use your driver
db.posts.createIndex({ "postPenguinId": 1 }, { unique: true })
db.posts.createIndex({ "slug": 1 }, { unique: true })
db.posts.createIndex({ "status": 1 })
db.posts.createIndex({ "publishedAt": -1 })
db.posts.createIndex({ "tags": 1 })

🚀 Node.js Webhook Handler

// webhook-handler.js
const express = require('express')
const { MongoClient } = require('mongodb')
const crypto = require('crypto')

const app = express()
app.use(express.json())

// MongoDB connection
const client = new MongoClient(process.env.MONGODB_URI)
let db

async function connectDB() {
    await client.connect()
    db = client.db('your_database')
    console.log('Connected to MongoDB')
}

// Verify webhook signature
function verifySignature(payload, signature, secret) {
    const expected = crypto
        .createHmac('sha256', secret)
        .update(payload)
        .digest('hex')
    const received = signature.replace('sha256=', '')
    return crypto.timingSafeEqual(Buffer.from(expected), Buffer.from(received))
}

// PostPenguin webhook endpoint
app.post('/api/webhooks/postpenguin', async (req, res) => {
    try {
        // Verify signature
        const signature = req.headers['x-postpenguin-signature']
        const secret = process.env.POSTPENGUIN_WEBHOOK_SECRET
        
        if (secret && signature) {
            const payload = JSON.stringify(req.body)
            if (!verifySignature(payload, signature, secret)) {
                return res.status(401).json({ error: 'Invalid signature' })
            }
        }
        
        const { 
            postPenguinId, 
            title, 
            slug, 
            html, 
            meta_title, 
            meta_description, 
            featured_image,
            tags 
        } = req.body
        
        // Validate required fields
        if (!title || !slug || !html) {
            return res.status(400).json({ error: 'Missing required fields' })
        }
        
        const collection = db.collection('posts')
        const now = new Date()
        
        // Upsert post (insert or update)
        const result = await collection.findOneAndUpdate(
            { postPenguinId: postPenguinId || `pp_${Date.now()}` },
            {
                $set: {
                    title,
                    slug,
                    html,
                    metaTitle: meta_title || title,
                    metaDescription: meta_description || '',
                    featuredImage: featured_image || null,
                    tags: tags || [],
                    status: 'publish',
                    publishedAt: now,
                    updatedAt: now
                },
                $setOnInsert: {
                    createdAt: now
                }
            },
            { 
                upsert: true, 
                returnDocument: 'after' 
            }
        )
        
        console.log(`✅ Post saved: ${title}`)
        
        res.status(200).json({
            success: true,
            postId: result._id.toString(),
            postPenguinId: result.postPenguinId
        })
        
    } catch (error) {
        console.error('Webhook error:', error)
        res.status(500).json({ error: 'Internal server error' })
    }
})

// Fetch posts API
app.get('/api/posts', async (req, res) => {
    try {
        const { slug, limit = 10, skip = 0, status = 'publish' } = req.query
        const collection = db.collection('posts')
        
        if (slug) {
            const post = await collection.findOne({ slug, status })
            if (!post) {
                return res.status(404).json({ error: 'Post not found' })
            }
            return res.json({ post })
        }
        
        const posts = await collection
            .find({ status })
            .sort({ publishedAt: -1 })
            .skip(parseInt(skip))
            .limit(parseInt(limit))
            .toArray()
        
        const total = await collection.countDocuments({ status })
        
        res.json({
            posts,
            pagination: {
                total,
                limit: parseInt(limit),
                skip: parseInt(skip)
            }
        })
        
    } catch (error) {
        console.error('Error fetching posts:', error)
        res.status(500).json({ error: 'Internal server error' })
    }
})

const PORT = process.env.PORT || 3001

connectDB().then(() => {
    app.listen(PORT, () => {
        console.log(`🚀 Server running on port ${PORT}`)
    })
})

🐍 Python with PyMongo

# webhook_handler.py
from flask import Flask, request, jsonify
from pymongo import MongoClient
import hmac
import hashlib
import os
from datetime import datetime

app = Flask(__name__)

# MongoDB connection
client = MongoClient(os.environ['MONGODB_URI'])
db = client['your_database']
posts = db['posts']

def verify_signature(payload, signature, secret):
    expected = hmac.new(
        secret.encode(),
        payload.encode(),
        hashlib.sha256
    ).hexdigest()
    received = signature.replace('sha256=', '')
    return hmac.compare_digest(expected, received)

@app.route('/api/webhooks/postpenguin', methods=['POST'])
def webhook():
    try:
        # Verify signature
        signature = request.headers.get('X-PostPenguin-Signature', '')
        secret = os.environ.get('POSTPENGUIN_WEBHOOK_SECRET', '')
        
        if secret and signature:
            if not verify_signature(request.get_data(as_text=True), signature, secret):
                return jsonify({'error': 'Invalid signature'}), 401
        
        data = request.json
        
        if not all(k in data for k in ['title', 'slug', 'html']):
            return jsonify({'error': 'Missing required fields'}), 400
        
        now = datetime.utcnow()
        
        result = posts.find_one_and_update(
            {'postPenguinId': data.get('postPenguinId', f'pp_{int(now.timestamp())}')},
            {
                '$set': {
                    'title': data['title'],
                    'slug': data['slug'],
                    'html': data['html'],
                    'metaTitle': data.get('meta_title', data['title']),
                    'metaDescription': data.get('meta_description', ''),
                    'featuredImage': data.get('featured_image'),
                    'tags': data.get('tags', []),
                    'status': 'publish',
                    'publishedAt': now,
                    'updatedAt': now
                },
                '$setOnInsert': {
                    'createdAt': now
                }
            },
            upsert=True,
            return_document=True
        )
        
        return jsonify({
            'success': True,
            'postId': str(result['_id'])
        })
        
    except Exception as e:
        print(f'Webhook error: {e}')
        return jsonify({'error': 'Internal server error'}), 500

if __name__ == '__main__':
    app.run(port=3001)

⚙️ Environment Variables

# .env
MONGODB_URI=mongodb+srv://user:[email protected]/database
POSTPENGUIN_WEBHOOK_SECRET=your-secret-key-here
PORT=3001

🧪 Testing

Test Webhook

curl -X POST http://localhost:3001/api/webhooks/postpenguin \
  -H "Content-Type: application/json" \
  -d '{
    "postPenguinId": "test_mongo_123",
    "title": "Test MongoDB Post",
    "slug": "test-mongodb-post",
    "html": "<p className="text-gray-700">This is a test post stored in MongoDB.</p>",
    "meta_title": "Test Post",
    "meta_description": "Testing MongoDB webhook",
    "tags": ["test", "mongodb"]
  }'

Verify in MongoDB

# Using mongosh
mongosh "mongodb+srv://cluster.mongodb.net/database"

# Find the test post
db.posts.findOne({ slug: "test-mongodb-post" })

# List recent posts
db.posts.find({ status: "publish" }).sort({ publishedAt: -1 }).limit(5)

📊 Useful Queries

// Get recent posts
db.posts.find({ status: "publish" })
    .sort({ publishedAt: -1 })
    .limit(10)

// Search by title (text search)
db.posts.createIndex({ title: "text", html: "text" })
db.posts.find({ $text: { $search: "keyword" } })

// Find posts by tag
db.posts.find({ tags: "mongodb" })

// Count posts by status
db.posts.aggregate([
    { $group: { _id: "$status", count: { $sum: 1 } } }
])

// Get post by slug
db.posts.findOne({ slug: "my-post-slug" })