feat(notion): implement Connector interface, markdown renderer, and DI registration

- Implement Validate, ListResources, FetchAll, FetchIncremental - Per-record database sync with individual knowledge items - Incremental sync with record-level change detection for databases - Tree-structured resource listing with parent-child relationships - Markdown renderer for all Notion block types including nested content - Register NotionConnector in DI container - Add unit tests for connector and markdown
2026-04-06 23:31:46 +08:00
parent 9f35abafff
commit 2cd979e1ea
6 changed files with 1992 additions and 1 deletions
@@ -513,6 +513,9 @@ func (s *DataSourceService) ProcessSync(ctx context.Context, task *asynq.Task) e
 	for _, item := range items {
 		if item.IsDeleted {
 			if ds.SyncDeletions {
+				// Count only — actual KB deletion is intentionally not performed.
+				// Users manage knowledge removal explicitly via the KB UI to avoid
+				// accidental data loss from connector misdetection or reconfiguration.
 				result.Deleted++
 			}
 			continue
@@ -52,6 +52,7 @@ import (
 	"github.com/Tencent/WeKnora/internal/database"
 	"github.com/Tencent/WeKnora/internal/datasource"
 	feishuConnector "github.com/Tencent/WeKnora/internal/datasource/connector/feishu"
+	notionConnector "github.com/Tencent/WeKnora/internal/datasource/connector/notion"
 	"github.com/Tencent/WeKnora/internal/event"
 	"github.com/Tencent/WeKnora/internal/handler"
 	"github.com/Tencent/WeKnora/internal/handler/session"
@@ -1277,8 +1278,10 @@ func initConnectorRegistry() *datasource.ConnectorRegistry {
 	// Register Feishu connector
 	_ = registry.Register(feishuConnector.NewConnector())

+	// Register Notion connector
+	_ = registry.Register(notionConnector.NewConnector())
+
 	// Future connectors will be registered here:
-	// _ = registry.Register(notionConnector.NewConnector())
 	// _ = registry.Register(confluenceConnector.NewConnector())
 	// _ = registry.Register(yuqueConnector.NewConnector())
 	// _ = registry.Register(githubConnector.NewConnector())
@@ -0,0 +1,852 @@
+package notion
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+	"sort"
+	"strings"
+	"time"
+
+	"github.com/Tencent/WeKnora/internal/logger"
+	"github.com/Tencent/WeKnora/internal/types"
+)
+
+const (
+	contentTypeMarkdown  = "text/markdown"
+	objectTypePage       = "page"
+	objectTypeDatabase   = "database"
+	objectTypeAttachment = "attachment"
+	defaultUntitledName  = "Untitled"
+)
+
+// Connector implements the datasource.Connector interface for Notion.
+type Connector struct{}
+
+// NewConnector creates a new Notion connector.
+func NewConnector() *Connector {
+	return &Connector{}
+}
+
+// Type returns the connector type identifier.
+func (c *Connector) Type() string {
+	return types.ConnectorTypeNotion
+}
+
+// Validate verifies that the Notion configuration is valid by testing connectivity.
+func (c *Connector) Validate(ctx context.Context, config *types.DataSourceConfig) error {
+	notionCfg, err := parseNotionConfig(config)
+	if err != nil {
+		return err
+	}
+
+	client := newClient(notionCfg.APIKey, extractBaseURL(config))
+	return client.Ping(ctx)
+}
+
+// ListResources lists all accessible Notion pages and databases as selectable resources.
+// Returns all objects with parent-child relationships populated, allowing the frontend
+// to render a tree view. Root objects have empty ParentID.
+func (c *Connector) ListResources(ctx context.Context, config *types.DataSourceConfig) ([]types.Resource, error) {
+	notionCfg, err := parseNotionConfig(config)
+	if err != nil {
+		return nil, err
+	}
+
+	client := newClient(notionCfg.APIKey, extractBaseURL(config))
+	pages, err := client.SearchPages(ctx)
+	if err != nil {
+		return nil, fmt.Errorf("search notion pages: %w", err)
+	}
+
+	allIDs := make(map[string]bool, len(pages))
+	for _, p := range pages {
+		allIDs[p.ID] = true
+	}
+
+	// Precompute effective parent for each page.
+	// data_source objects need database_parent (their `parent` points at the
+	// database container, not the workspace location); see resolveParentID.
+	parentOf := make(map[string]string, len(pages))
+	for _, p := range pages {
+		if p.InTrash {
+			continue
+		}
+		parentOf[p.ID] = resolveParentID(p, allIDs)
+	}
+
+	childrenCount := make(map[string]int)
+	for _, pid := range parentOf {
+		if pid != "" {
+			childrenCount[pid]++
+		}
+	}
+
+	var resources []types.Resource
+	for _, p := range pages {
+		if p.InTrash {
+			continue
+		}
+
+		resourceType := objectTypePage
+		if p.isDatabase() {
+			resourceType = objectTypeDatabase
+		}
+
+		resources = append(resources, types.Resource{
+			ExternalID:  p.ID,
+			Name:        p.Title,
+			Type:        resourceType,
+			URL:         p.URL,
+			ParentID:    parentOf[p.ID],
+			HasChildren: childrenCount[p.ID] > 0,
+		})
+	}
+
+	return resources, nil
+}
+
+// FetchAll performs a full sync of all documents from the specified resources.
+func (c *Connector) FetchAll(ctx context.Context, config *types.DataSourceConfig, resourceIDs []string) ([]types.FetchedItem, error) {
+	notionCfg, err := parseNotionConfig(config)
+	if err != nil {
+		return nil, err
+	}
+
+	client := newClient(notionCfg.APIKey, extractBaseURL(config))
+	visited := c.excludedSetFromListResources(ctx, config, resourceIDs)
+	var allItems []types.FetchedItem
+
+	for _, resourceID := range resourceIDs {
+		page, err := client.GetPage(ctx, resourceID)
+		if err == nil {
+			allItems = append(allItems, c.fetchPage(ctx, client, page, visited)...)
+			continue
+		}
+		// Not a page — treat as database/data_source.
+		// fetchDatabase handles both data_source IDs (from search) and
+		// database container IDs (from child_database blocks).
+		items := c.fetchDatabase(ctx, client, resourceID, visited)
+		if len(items) == 0 {
+			logger.Warnf(ctx, "[Notion] failed to fetch resource %s as page or database", resourceID)
+		}
+		allItems = append(allItems, items...)
+	}
+
+	return allItems, nil
+}
+
+// FetchIncremental performs an incremental sync based on cursor state.
+// On first sync (nil cursor), delegates to FetchAll for efficiency (no discovery needed).
+func (c *Connector) FetchIncremental(ctx context.Context, config *types.DataSourceConfig, cursor *types.SyncCursor) ([]types.FetchedItem, *types.SyncCursor, error) {
+	notionCfg, err := parseNotionConfig(config)
+	if err != nil {
+		return nil, nil, err
+	}
+
+	resourceIDs := config.ResourceIDs
+	if len(resourceIDs) == 0 {
+		return nil, nil, fmt.Errorf("no resource IDs configured")
+	}
+
+	client := newClient(notionCfg.APIKey, extractBaseURL(config))
+
+	// Parse previous cursor
+	var prevCursor notionCursor
+	if cursor != nil && cursor.ConnectorCursor != nil {
+		cursorBytes, _ := json.Marshal(cursor.ConnectorCursor)
+		_ = json.Unmarshal(cursorBytes, &prevCursor)
+	}
+
+	// First sync: use FetchAll then build cursor from Search API (one call, no block traversal)
+	isFirstSync := len(prevCursor.PageEditTimes) == 0
+	if isFirstSync {
+		logger.Infof(ctx, "[Notion] first sync, using FetchAll for %d resources", len(resourceIDs))
+		items, err := c.FetchAll(ctx, config, resourceIDs)
+		if err != nil {
+			return nil, nil, err
+		}
+
+		// Build cursor from fetched items' UpdatedAt timestamps.
+		// Record-level edit times are tracked individually (object_type == "page").
+		// Database container IDs are also added so incremental sync can detect
+		// whether a database actually changed, avoiding full record queries every cycle.
+		newEditTimes := make(map[string]time.Time)
+		for _, item := range items {
+			if item.Metadata["object_type"] == objectTypePage {
+				newEditTimes[item.ExternalID] = item.UpdatedAt
+			}
+		}
+		// Ensure all selected resourceIDs appear in the cursor.
+		// Pages already appear via their items; databases need explicit entries.
+		for _, rid := range resourceIDs {
+			if _, ok := newEditTimes[rid]; !ok {
+				newEditTimes[rid] = time.Now()
+			}
+		}
+
+		return items, buildCursor(newEditTimes), nil
+	}
+
+	// Subsequent syncs: discover all pages, diff against cursor, fetch only changes
+	logger.Infof(ctx, "[Notion] incremental sync, discovering pages")
+	pages, fetchVisited := c.discoverAllResources(ctx, client, resourceIDs)
+	logger.Infof(ctx, "[Notion] discovered %d pages", len(pages))
+
+	newEditTimes := make(map[string]time.Time)
+	pageByID := make(map[string]*notionPage, len(pages))
+	for i := range pages {
+		newEditTimes[pages[i].ID] = pages[i].LastEditedTime
+		pageByID[pages[i].ID] = &pages[i]
+	}
+
+	var changedItems []types.FetchedItem
+	changedCount := 0
+
+	for pageID, newTime := range newEditTimes {
+		prevTime, existed := prevCursor.PageEditTimes[pageID]
+		if existed && newTime.Equal(prevTime) {
+			continue
+		}
+		if fetchVisited[pageID] {
+			continue
+		}
+		changedCount++
+		pg := pageByID[pageID]
+		if pg == nil {
+			continue
+		}
+		logger.Debugf(ctx, "[Notion] changed: %s (%s, %s)", pg.Title, pg.ID, pg.Object)
+		if pg.isDatabase() {
+			// Incremental database sync: query records, diff against cursor,
+			// only fetch blocks for records whose edit time actually changed.
+			items, recordEditTimes := c.fetchDatabaseIncremental(ctx, client, pg.ID, prevCursor.PageEditTimes, fetchVisited)
+			changedItems = append(changedItems, items...)
+			// Merge record-level edit times into the cursor
+			for rid, rt := range recordEditTimes {
+				newEditTimes[rid] = rt
+			}
+		} else {
+			changedItems = append(changedItems, c.fetchPage(ctx, client, pg, fetchVisited)...)
+		}
+	}
+
+	// Detect deletions. Only emit IsDeleted for pages that previously belonged
+	// to a selected resource and have actually disappeared at source. Pages that
+	// are still visible but no longer reachable from any selected root (i.e. the
+	// user deselected their ancestor) live in fetchVisited as the "excluded" set
+	// returned by discoverAllResources, and must NOT be reported as deletions.
+	for pageID := range prevCursor.PageEditTimes {
+		if _, exists := newEditTimes[pageID]; exists {
+			continue
+		}
+		if fetchVisited[pageID] {
+			continue
+		}
+		changedItems = append(changedItems, types.FetchedItem{
+			ExternalID: pageID,
+			IsDeleted:  true,
+			Metadata:   map[string]string{"channel": types.ChannelNotion},
+		})
+	}
+
+	logger.Infof(ctx, "[Notion] incremental: %d changed, %d total items", changedCount, len(changedItems))
+
+	return changedItems, buildCursor(newEditTimes), nil
+}
+
+func buildCursor(editTimes map[string]time.Time) *types.SyncCursor {
+	now := time.Now()
+	cursorData := notionCursor{
+		PageEditTimes: editTimes,
+	}
+	cursorBytes, _ := json.Marshal(cursorData)
+	var cursorMap map[string]interface{}
+	_ = json.Unmarshal(cursorBytes, &cursorMap)
+
+	return &types.SyncCursor{
+		LastSyncTime:    now,
+		ConnectorCursor: cursorMap,
+	}
+}
+
+// --- Internal methods ---
+
+// fetchPage fetches a single page's content and attachments.
+// If page is nil, it will be fetched from the API.
+func (c *Connector) fetchPage(ctx context.Context, client *notionClient, page *notionPage, visited map[string]bool) []types.FetchedItem {
+	if page == nil {
+		return nil
+	}
+	if visited[page.ID] {
+		return nil
+	}
+	visited[page.ID] = true
+
+	if page.InTrash {
+		return nil
+	}
+
+	// Database records store content in properties, not blocks — delegate to
+	// buildRecordItem so they aren't dropped as empty pages. Record parent.type
+	// can be either "database_id" (older GetPage responses) or "data_source_id"
+	// (search API and 2025-09-03+ workspaces); accept both.
+	if page.Parent.Type == parentTypeDatabaseID || page.Parent.Type == parentTypeDataSourceID {
+		dbTitle := ""
+		if dbInfo, err := getDatabaseOrDataSourceInfo(ctx, client, page.Parent.GetParentID()); err != nil {
+			logger.Debugf(ctx, "[Notion] failed to resolve parent db title for record %s: %v", page.ID, err)
+		} else {
+			dbTitle = dbInfo.Page.Title
+		}
+		propNames := extractPropertySchema(*page)
+		if item := c.buildRecordItem(ctx, client, *page, propNames, dbTitle); item != nil {
+			return []types.FetchedItem{*item}
+		}
+		return nil
+	}
+
+	blocks, err := client.GetBlockChildrenAll(ctx, page.ID)
+	if err != nil {
+		logger.Warnf(ctx, "[Notion] failed to get blocks for page %s: %v", page.ID, err)
+		return nil
+	}
+
+	resolveFileUploads(ctx, client, blocks)
+
+	markdown, attachmentList := BlocksToMarkdown(blocks)
+
+	var items []types.FetchedItem
+
+	// Only skip truly empty pages (no content at all)
+	if strings.TrimSpace(markdown) != "" {
+		fileName := page.Title + ".md"
+		if page.Title == "" {
+			fileName = defaultUntitledName + ".md"
+		}
+
+		items = append(items, types.FetchedItem{
+			ExternalID:  page.ID,
+			Title:       page.Title,
+			Content:     []byte(markdown),
+			ContentType: contentTypeMarkdown,
+			FileName:    fileName,
+			URL:         page.URL,
+			UpdatedAt:   page.LastEditedTime,
+			Metadata: map[string]string{
+				"channel":     types.ChannelNotion,
+				"object_type": objectTypePage,
+			},
+		})
+	}
+
+	// Download file attachments (PDF, documents, etc.)
+	// Skip images — they're already referenced in Markdown as ![](url),
+	// and downloading them as separate items requires VLM to process.
+	for _, att := range attachmentList {
+		if att.URL == "" || att.Type == "image" {
+			continue
+		}
+		data, err := client.DownloadFile(ctx, att.URL)
+		if err != nil {
+			logger.Warnf(ctx, "[Notion] failed to download attachment %s: %v", att.FileName, err)
+			continue
+		}
+		items = append(items, types.FetchedItem{
+			ExternalID:       fmt.Sprintf("%s:%s", page.ID, att.FileName),
+			Title:            att.FileName,
+			Content:          data,
+			ContentType:      mimeTypeForAttachment(att.Type),
+			FileName:         att.FileName,
+			SourceResourceID: page.ID,
+			Metadata: map[string]string{
+				"channel":     types.ChannelNotion,
+				"object_type": objectTypeAttachment,
+			},
+		})
+	}
+
+	for _, block := range blocks {
+		switch block.Type {
+		case "child_page":
+			childPage, err := client.GetPage(ctx, block.ID)
+			if err != nil {
+				logger.Warnf(ctx, "[Notion] failed to get child page %s: %v", block.ID, err)
+				continue
+			}
+			items = append(items, c.fetchPage(ctx, client, childPage, visited)...)
+		case "child_database":
+			items = append(items, c.fetchDatabase(ctx, client, block.ID, visited)...)
+		}
+	}
+
+	return items
+}
+
+// fetchDatabase syncs each database record as an individual knowledge item (full sync).
+// Accepts either a data_source_id (from search) or database_id (from child_database blocks).
+func (c *Connector) fetchDatabase(ctx context.Context, client *notionClient, id string, visited map[string]bool) []types.FetchedItem {
+	if visited[id] {
+		return nil
+	}
+	visited[id] = true
+
+	records, dbTitle, queryID, err := c.queryDatabaseRecords(ctx, client, id)
+	if err != nil || len(records) == 0 {
+		return nil
+	}
+	// Same database may be reached via two different IDs (data_source_id from search
+	// vs database_container_id from child_database blocks). Mark the canonical
+	// data_source_id as visited so the second path is deduped.
+	if queryID != "" && queryID != id {
+		if visited[queryID] {
+			return nil
+		}
+		visited[queryID] = true
+	}
+
+	propNames := extractPropertySchema(records[0])
+	var items []types.FetchedItem
+	for _, record := range records {
+		if record.InTrash {
+			continue
+		}
+		item := c.buildRecordItem(ctx, client, record, propNames, dbTitle)
+		if item != nil {
+			items = append(items, *item)
+		}
+	}
+	return items
+}
+
+// fetchDatabaseIncremental syncs only changed records by comparing edit times against cursor.
+// Returns fetched items and a map of record_id → edit_time for cursor update.
+func (c *Connector) fetchDatabaseIncremental(ctx context.Context, client *notionClient, id string, prevEditTimes map[string]time.Time, visited map[string]bool) ([]types.FetchedItem, map[string]time.Time) {
+	if visited[id] {
+		return nil, nil
+	}
+	visited[id] = true
+
+	records, dbTitle, queryID, err := c.queryDatabaseRecords(ctx, client, id)
+	if err != nil {
+		return nil, nil
+	}
+	if queryID != "" && queryID != id {
+		if visited[queryID] {
+			return nil, nil
+		}
+		visited[queryID] = true
+	}
+
+	recordEditTimes := make(map[string]time.Time, len(records))
+	var propNames []string
+	if len(records) > 0 {
+		propNames = extractPropertySchema(records[0])
+	}
+
+	var items []types.FetchedItem
+	changedCount := 0
+	for _, record := range records {
+		if record.InTrash {
+			continue
+		}
+		recordEditTimes[record.ID] = record.LastEditedTime
+
+		prevTime, existed := prevEditTimes[record.ID]
+		if existed && record.LastEditedTime.Equal(prevTime) {
+			continue
+		}
+		changedCount++
+		item := c.buildRecordItem(ctx, client, record, propNames, dbTitle)
+		if item != nil {
+			items = append(items, *item)
+		}
+	}
+
+	logger.Infof(ctx, "[Notion] database %s incremental: %d changed out of %d records", id, changedCount, len(records))
+	return items, recordEditTimes
+}
+
+// queryDatabaseRecords resolves the database ID and queries all records.
+// Returns the records, the database title, and the canonical data_source_id used for the query.
+func (c *Connector) queryDatabaseRecords(ctx context.Context, client *notionClient, id string) ([]notionPage, string, string, error) {
+	dbInfo, err := getDatabaseOrDataSourceInfo(ctx, client, id)
+	if err != nil {
+		logger.Warnf(ctx, "[Notion] failed to get database/data_source info %s: %v", id, err)
+		return nil, "", "", err
+	}
+
+	queryID := dbInfo.DataSourceID
+	if queryID == "" {
+		queryID = id
+	}
+	records, err := client.QueryDatabaseAll(ctx, queryID)
+	if err != nil {
+		logger.Warnf(ctx, "[Notion] failed to query database %s: %v", id, err)
+		return nil, "", "", err
+	}
+
+	logger.Infof(ctx, "[Notion] database %s (%s): %d records", id, dbInfo.Page.Title, len(records))
+	return records, dbInfo.Page.Title, queryID, nil
+}
+
+// buildRecordItem converts a single database record into a FetchedItem
+// with properties as header and block content as body.
+func (c *Connector) buildRecordItem(ctx context.Context, client *notionClient, record notionPage, propNames []string, dbTitle string) *types.FetchedItem {
+	var content strings.Builder
+	title := record.Title
+	if title == "" {
+		title = defaultUntitledName
+	}
+	content.WriteString("# " + title + "\n\n")
+
+	if record.RawProperties != nil {
+		var props map[string]interface{}
+		json.Unmarshal(record.RawProperties, &props)
+		for _, name := range propNames {
+			if propMap, ok := props[name].(map[string]interface{}); ok {
+				val := propertyToString(propMap)
+				if val != "" {
+					content.WriteString("- **" + name + "**: " + val + "\n")
+				}
+			}
+		}
+	}
+
+	// Database records may contain page-style block content beyond their properties
+	blocks, err := client.GetBlockChildrenAll(ctx, record.ID)
+	if err != nil {
+		logger.Warnf(ctx, "[Notion] failed to get blocks for record %s: %v", record.ID, err)
+	} else if len(blocks) > 0 {
+		resolveFileUploads(ctx, client, blocks)
+		markdown, _ := BlocksToMarkdown(blocks)
+		if strings.TrimSpace(markdown) != "" {
+			content.WriteString("\n" + markdown)
+		}
+	}
+
+	bodyStr := strings.TrimSpace(content.String())
+	if bodyStr == "" {
+		return nil
+	}
+
+	return &types.FetchedItem{
+		ExternalID:  record.ID,
+		Title:       title,
+		Content:     []byte(bodyStr),
+		ContentType: contentTypeMarkdown,
+		FileName:    title + ".md",
+		URL:         record.URL,
+		UpdatedAt:   record.LastEditedTime,
+		Metadata: map[string]string{
+			"channel":     types.ChannelNotion,
+			"object_type": objectTypePage,
+			"database":    dbTitle,
+		},
+	}
+}
+
+// discoverAllResources uses the Search API to find all pages and data_sources,
+// then filters by parent chain via BFS from the selected resource IDs.
+// Returns the included pages plus the excluded set (visible pages NOT under any
+// selected root) so callers can seed `visited` for child-block recursion without
+// a second SearchPages round-trip.
+func (c *Connector) discoverAllResources(ctx context.Context, client *notionClient, resourceIDs []string) (included []notionPage, excluded map[string]bool) {
+	allPages, err := client.SearchPages(ctx)
+	if err != nil {
+		logger.Warnf(ctx, "[Notion] failed to search pages for discovery: %v", err)
+		return nil, nil
+	}
+
+	// data_source objects use database_parent for workspace hierarchy.
+	allIDs := make(map[string]bool, len(allPages))
+	pageByID := make(map[string]notionPage, len(allPages))
+	for _, p := range allPages {
+		allIDs[p.ID] = true
+		pageByID[p.ID] = p
+	}
+
+	childrenOf := make(map[string][]string)
+	for _, p := range allPages {
+		if p.InTrash {
+			continue
+		}
+		parentID := resolveParentID(p, allIDs)
+		if parentID != "" {
+			childrenOf[parentID] = append(childrenOf[parentID], p.ID)
+		}
+	}
+
+	// BFS from each resource root to collect all descendants
+	includedSet := make(map[string]bool)
+	queue := make([]string, 0, len(resourceIDs))
+	for _, id := range resourceIDs {
+		if _, ok := pageByID[id]; ok {
+			includedSet[id] = true
+			queue = append(queue, id)
+		}
+	}
+	for len(queue) > 0 {
+		current := queue[0]
+		queue = queue[1:]
+		for _, childID := range childrenOf[current] {
+			if !includedSet[childID] {
+				includedSet[childID] = true
+				queue = append(queue, childID)
+			}
+		}
+	}
+
+	included = make([]notionPage, 0, len(includedSet))
+	for id := range includedSet {
+		included = append(included, pageByID[id])
+	}
+	excluded = make(map[string]bool, len(allPages)-len(includedSet))
+	for id := range allIDs {
+		if !includedSet[id] {
+			excluded[id] = true
+		}
+	}
+	return included, excluded
+}
+
+// --- File upload resolution ---
+
+// resolveFileUploads walks the block tree and resolves file_upload type files
+// by re-fetching the block to get a temporary download URL.
+// This mutates blocks in place, replacing file_upload RawContent with resolved URLs.
+func resolveFileUploads(ctx context.Context, client *notionClient, blocks []notionBlock) {
+	for i := range blocks {
+		if isFileBlock(blocks[i].Type) && blocks[i].RawContent != nil {
+			var file notionFile
+			json.Unmarshal(blocks[i].RawContent, &file)
+			if file.GetFileUploadID() != "" {
+				resolved, err := client.ResolveBlock(ctx, blocks[i].ID)
+				if err != nil {
+					logger.Warnf(ctx, "[Notion] failed to resolve file_upload in block %s: %v", blocks[i].ID, err)
+					continue
+				}
+				blocks[i].RawContent = resolved.RawContent
+			}
+		}
+		if len(blocks[i].Children) > 0 {
+			resolveFileUploads(ctx, client, blocks[i].Children)
+		}
+	}
+}
+
+func isFileBlock(blockType string) bool {
+	switch blockType {
+	case "image", "file", "pdf", "video", "audio":
+		return true
+	}
+	return false
+}
+
+// --- Property extraction ---
+
+// propertyToString recursively extracts a string value from a Notion property.
+// Follows the type chain generically, handling all 22 property types without hardcoding.
+func propertyToString(value map[string]interface{}) string {
+	if value == nil {
+		return ""
+	}
+
+	typeName, _ := value["type"].(string)
+	if typeName == "" {
+		return extractLeafValue(value)
+	}
+
+	inner, exists := value[typeName]
+	if !exists || inner == nil {
+		return ""
+	}
+
+	return extractValue(inner)
+}
+
+func extractValue(v interface{}) string {
+	switch val := v.(type) {
+	case string:
+		return val
+	case float64:
+		if val == float64(int64(val)) {
+			return fmt.Sprintf("%d", int64(val))
+		}
+		return fmt.Sprintf("%g", val)
+	case bool:
+		return fmt.Sprintf("%t", val)
+	case map[string]interface{}:
+		return extractLeafValue(val)
+	case []interface{}:
+		var parts []string
+		for _, item := range val {
+			s := extractValue(item)
+			if s != "" {
+				parts = append(parts, s)
+			}
+		}
+		return strings.Join(parts, ", ")
+	case nil:
+		return ""
+	default:
+		return fmt.Sprint(v)
+	}
+}
+
+func extractLeafValue(m map[string]interface{}) string {
+	if name, ok := m["name"].(string); ok {
+		return name
+	}
+	if content, ok := m["content"].(string); ok {
+		return content
+	}
+	if pt, ok := m["plain_text"].(string); ok {
+		return pt
+	}
+	if start, ok := m["start"].(string); ok {
+		if end, ok := m["end"].(string); ok && end != "" {
+			return start + " ~ " + end
+		}
+		return start
+	}
+	if expr, ok := m["expression"].(string); ok {
+		return expr
+	}
+	// Recurse if there's a type field
+	if typeName, ok := m["type"].(string); ok {
+		if inner, ok := m[typeName]; ok {
+			return extractValue(inner)
+		}
+	}
+	return ""
+}
+
+// extractPropertySchema extracts non-title property names from a record.
+// Names are sorted alphabetically to ensure deterministic order
+// (prevents false-positive content changes on incremental sync due to map iteration randomness).
+func extractPropertySchema(record notionPage) (propNames []string) {
+	if record.RawProperties == nil {
+		return nil
+	}
+	var props map[string]json.RawMessage
+	if err := json.Unmarshal(record.RawProperties, &props); err != nil {
+		return nil
+	}
+	for name, propRaw := range props {
+		var prop struct {
+			Type string `json:"type"`
+		}
+		json.Unmarshal(propRaw, &prop)
+		if prop.Type != "title" {
+			propNames = append(propNames, name)
+		}
+	}
+	sort.Strings(propNames)
+	return
+}
+
+// resolveParentID determines where a page/data_source sits in the workspace hierarchy.
+// For data_source objects, uses database_parent which points directly at the
+// containing page; the regular `parent` field on a data_source points at its
+// database container, not at the workspace location, so it's the wrong source.
+func resolveParentID(p notionPage, allIDs map[string]bool) string {
+	// Data sources: use database_parent which shows the workspace location
+	// (e.g., which page contains this database), not the database container ID.
+	if p.isDatabase() && p.DatabaseParent != nil {
+		pid := p.DatabaseParent.GetParentID()
+		if pid != "" && allIDs[pid] {
+			return pid
+		}
+		return ""
+	}
+	// Pages: use direct parent
+	if p.Parent.Type == parentTypeWorkspace {
+		return ""
+	}
+	pid := p.Parent.GetParentID()
+	if pid != "" && allIDs[pid] {
+		return pid
+	}
+	return ""
+}
+
+// getDatabaseOrDataSourceInfo retrieves metadata for a database or data_source.
+// Tries GET /v1/data_sources/{id} first (search returns data_source IDs),
+// falls back to GET /v1/databases/{id} (child_database blocks use database IDs).
+func getDatabaseOrDataSourceInfo(ctx context.Context, client *notionClient, id string) (*databaseInfo, error) {
+	ds, err := client.GetDataSourceInfo(ctx, id)
+	if err == nil {
+		return &databaseInfo{Page: *ds, DataSourceID: id}, nil
+	}
+	return client.GetDatabaseInfo(ctx, id)
+}
+
+// computeExcludedSet returns IDs the user has explicitly deselected: visible in
+// the picker but neither selected nor a descendant of a selected node. Used to
+// seed `visited` so recursive child_page/child_database traversal skips them.
+// Pages the user has never seen (e.g. added after the last config save) are
+// not excluded, so a selected parent still picks them up automatically.
+func computeExcludedSet(visibleIDs []string, parentOf map[string]string, selectedIDs []string) map[string]bool {
+	selected := make(map[string]bool, len(selectedIDs))
+	for _, id := range selectedIDs {
+		selected[id] = true
+	}
+	excluded := make(map[string]bool, len(visibleIDs))
+	for _, id := range visibleIDs {
+		hasSelectedAncestor := false
+		for cur := id; cur != ""; cur = parentOf[cur] {
+			if selected[cur] {
+				hasSelectedAncestor = true
+				break
+			}
+		}
+		if !hasSelectedAncestor {
+			excluded[id] = true
+		}
+	}
+	return excluded
+}
+
+// excludedSetFromListResources fetches the picker hierarchy via ListResources
+// and computes the deselected set. Used by FetchAll where no other code path
+// already has the page list in hand.
+func (c *Connector) excludedSetFromListResources(ctx context.Context, config *types.DataSourceConfig, selectedIDs []string) map[string]bool {
+	visible, err := c.ListResources(ctx, config)
+	if err != nil {
+		logger.Warnf(ctx, "[Notion] failed to list visible resources for exclusion: %v", err)
+		return map[string]bool{}
+	}
+	ids := make([]string, 0, len(visible))
+	parentOf := make(map[string]string, len(visible))
+	for _, r := range visible {
+		ids = append(ids, r.ExternalID)
+		parentOf[r.ExternalID] = r.ParentID
+	}
+	return computeExcludedSet(ids, parentOf, selectedIDs)
+}
+
+// --- Helpers ---
+
+func extractBaseURL(config *types.DataSourceConfig) string {
+	if config.Settings != nil {
+		if url, ok := config.Settings["base_url"].(string); ok && url != "" {
+			return url
+		}
+	}
+	return DefaultBaseURL
+}
+
+func mimeTypeForAttachment(attType string) string {
+	switch attType {
+	case "image":
+		return "image/png"
+	case "pdf":
+		return "application/pdf"
+	case "video":
+		return "video/mp4"
+	case "audio":
+		return "audio/mpeg"
+	default:
+		return "application/octet-stream"
+	}
+}
@@ -0,0 +1,284 @@
+package notion
+
+import (
+	"context"
+	"testing"
+
+	"github.com/Tencent/WeKnora/internal/types"
+)
+
+func makeNotionConfig(cfg *Config, baseURL string, resourceIDs []string) *types.DataSourceConfig {
+	return &types.DataSourceConfig{
+		Type: types.ConnectorTypeNotion,
+		Credentials: map[string]interface{}{
+			"api_key": cfg.APIKey,
+		},
+		ResourceIDs: resourceIDs,
+		Settings: map[string]interface{}{
+			"base_url": baseURL,
+		},
+	}
+}
+
+func TestConnectorType(t *testing.T) {
+	c := NewConnector()
+	if c.Type() != types.ConnectorTypeNotion {
+		t.Errorf("Type() = %q, want %q", c.Type(), types.ConnectorTypeNotion)
+	}
+}
+
+func TestConnectorValidate(t *testing.T) {
+	ts, cfg := fakeNotion()
+	defer ts.Close()
+
+	c := NewConnector()
+	err := c.Validate(context.Background(), makeNotionConfig(cfg, ts.URL, nil))
+	if err != nil {
+		t.Fatalf("Validate() error: %v", err)
+	}
+}
+
+func TestConnectorValidate_BadToken(t *testing.T) {
+	ts, _ := fakeNotion()
+	defer ts.Close()
+
+	c := NewConnector()
+	err := c.Validate(context.Background(), makeNotionConfig(
+		&Config{APIKey: "bad-token"}, ts.URL, nil,
+	))
+	if err == nil {
+		t.Fatal("expected error for bad token")
+	}
+}
+
+func TestConnectorListResources(t *testing.T) {
+	ts, cfg := fakeNotion()
+	defer ts.Close()
+
+	c := NewConnector()
+	resources, err := c.ListResources(context.Background(), makeNotionConfig(cfg, ts.URL, nil))
+	if err != nil {
+		t.Fatalf("ListResources() error: %v", err)
+	}
+
+	if len(resources) != 2 {
+		t.Fatalf("expected 2 resources, got %d", len(resources))
+	}
+	if resources[0].ExternalID != "page-1" || resources[0].Type != "page" || resources[0].Name != "Test Page" {
+		t.Errorf("resource[0] = %+v", resources[0])
+	}
+	if resources[1].ExternalID != "db-1" || resources[1].Type != "database" || resources[1].Name != "Test Database" {
+		t.Errorf("resource[1] = %+v, want Name=%q", resources[1], "Test Database")
+	}
+}
+
+func TestConnectorFetchAll(t *testing.T) {
+	ts, cfg := fakeNotion()
+	defer ts.Close()
+
+	c := NewConnector()
+	items, err := c.FetchAll(context.Background(), makeNotionConfig(cfg, ts.URL, []string{"page-1"}), []string{"page-1"})
+	if err != nil {
+		t.Fatalf("FetchAll() error: %v", err)
+	}
+
+	if len(items) == 0 {
+		t.Fatal("expected at least 1 item")
+	}
+
+	// First item should be the page with markdown content
+	found := false
+	for _, item := range items {
+		if item.ExternalID == "page-1" {
+			found = true
+			if item.ContentType != "text/markdown" {
+				t.Errorf("ContentType = %q", item.ContentType)
+			}
+			if item.Metadata["channel"] != "notion" {
+				t.Errorf("channel = %q", item.Metadata["channel"])
+			}
+			if len(item.Content) == 0 {
+				t.Error("expected non-empty content")
+			}
+		}
+	}
+	if !found {
+		t.Error("page-1 not found in items")
+	}
+}
+
+func TestConnectorFetchAll_Database(t *testing.T) {
+	ts, cfg := fakeNotion()
+	defer ts.Close()
+
+	c := NewConnector()
+	items, err := c.FetchAll(context.Background(), makeNotionConfig(cfg, ts.URL, []string{"db-1"}), []string{"db-1"})
+	if err != nil {
+		t.Fatalf("FetchAll() error: %v", err)
+	}
+
+	if len(items) == 0 {
+		t.Fatal("expected at least 1 item")
+	}
+
+	// Each database record should be synced as an individual knowledge item
+	found := false
+	for _, item := range items {
+		if item.ExternalID == "record-1" {
+			found = true
+			if item.Metadata["object_type"] != "page" {
+				t.Errorf("object_type = %q, want %q", item.Metadata["object_type"], "page")
+			}
+			if item.Metadata["database"] != "Test Database" {
+				t.Errorf("database = %q, want %q", item.Metadata["database"], "Test Database")
+			}
+			if item.ContentType != "text/markdown" {
+				t.Errorf("ContentType = %q", item.ContentType)
+			}
+			if len(item.Content) == 0 {
+				t.Error("expected non-empty content")
+			}
+		}
+	}
+	if !found {
+		t.Error("record-1 not found in items")
+	}
+}
+
+// TestConnectorFetchAll_SingleRecord verifies that selecting a single database
+// row by ID routes through fetchPage's record-detection branch and produces an
+// item via buildRecordItem (instead of being silently dropped as an empty page).
+func TestConnectorFetchAll_SingleRecord(t *testing.T) {
+	ts, cfg := fakeNotion()
+	defer ts.Close()
+
+	c := NewConnector()
+	items, err := c.FetchAll(context.Background(), makeNotionConfig(cfg, ts.URL, []string{"record-1"}), []string{"record-1"})
+	if err != nil {
+		t.Fatalf("FetchAll() error: %v", err)
+	}
+	if len(items) != 1 {
+		t.Fatalf("expected 1 item, got %d", len(items))
+	}
+	if items[0].ExternalID != "record-1" {
+		t.Errorf("ExternalID = %q, want record-1", items[0].ExternalID)
+	}
+	if items[0].Metadata["object_type"] != "page" {
+		t.Errorf("object_type = %q, want page", items[0].Metadata["object_type"])
+	}
+	if len(items[0].Content) == 0 {
+		t.Error("expected non-empty content from buildRecordItem")
+	}
+}
+
+func TestConnectorFetchIncremental_NoChanges(t *testing.T) {
+	ts, cfg := fakeNotion()
+	defer ts.Close()
+
+	c := NewConnector()
+	config := makeNotionConfig(cfg, ts.URL, []string{"page-1"})
+
+	// First: full fetch to establish baseline
+	_, err := c.FetchAll(context.Background(), config, []string{"page-1"})
+	if err != nil {
+		t.Fatalf("FetchAll() error: %v", err)
+	}
+
+	// Build a cursor that matches current state
+	cursor := &types.SyncCursor{
+		ConnectorCursor: map[string]interface{}{
+			"page_edit_times": map[string]interface{}{
+				"page-1": "2026-01-15T10:00:00Z",
+			},
+		},
+	}
+
+	items, newCursor, err := c.FetchIncremental(context.Background(), config, cursor)
+	if err != nil {
+		t.Fatalf("FetchIncremental() error: %v", err)
+	}
+
+	// No changes expected (timestamps match)
+	if len(items) != 0 {
+		t.Errorf("expected 0 items for no changes, got %d", len(items))
+	}
+	if newCursor == nil {
+		t.Fatal("expected non-nil cursor")
+	}
+}
+
+func TestPropertyToString(t *testing.T) {
+	tests := []struct {
+		name     string
+		value    map[string]interface{}
+		expected string
+	}{
+		{
+			name: "select",
+			value: map[string]interface{}{
+				"type":   "select",
+				"select": map[string]interface{}{"name": "Done"},
+			},
+			expected: "Done",
+		},
+		{
+			name: "rich_text",
+			value: map[string]interface{}{
+				"type": "rich_text",
+				"rich_text": []interface{}{
+					map[string]interface{}{"plain_text": "Hello"},
+				},
+			},
+			expected: "Hello",
+		},
+		{
+			name: "number",
+			value: map[string]interface{}{
+				"type":   "number",
+				"number": 42.0,
+			},
+			expected: "42",
+		},
+		{
+			name: "checkbox true",
+			value: map[string]interface{}{
+				"type":     "checkbox",
+				"checkbox": true,
+			},
+			expected: "true",
+		},
+		{
+			name: "date",
+			value: map[string]interface{}{
+				"type": "date",
+				"date": map[string]interface{}{"start": "2026-01-15", "end": "2026-01-20"},
+			},
+			expected: "2026-01-15 ~ 2026-01-20",
+		},
+		{
+			name: "multi_select",
+			value: map[string]interface{}{
+				"type": "multi_select",
+				"multi_select": []interface{}{
+					map[string]interface{}{"name": "Tag1"},
+					map[string]interface{}{"name": "Tag2"},
+				},
+			},
+			expected: "Tag1, Tag2",
+		},
+		{
+			name:     "nil value",
+			value:    map[string]interface{}{"type": "rich_text", "rich_text": nil},
+			expected: "",
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			got := propertyToString(tt.value)
+			if got != tt.expected {
+				t.Errorf("propertyToString() = %q, want %q", got, tt.expected)
+			}
+		})
+	}
+}
@@ -0,0 +1,497 @@
+package notion
+
+import (
+	"encoding/json"
+	"fmt"
+	"strings"
+)
+
+// BlocksToMarkdown converts a tree of Notion blocks into Markdown text,
+// and collects attachment references (image, file, pdf, video, audio).
+func BlocksToMarkdown(blocks []notionBlock) (string, []attachment) {
+	var b strings.Builder
+	var attachments []attachment
+	renderBlocks(&b, blocks, 0, &attachments)
+	// Collapse 3+ consecutive newlines into 2 (avoids excessive blank lines from empty paragraphs)
+	result := b.String()
+	for strings.Contains(result, "\n\n\n") {
+		result = strings.ReplaceAll(result, "\n\n\n", "\n\n")
+	}
+	return strings.TrimSpace(result) + "\n", attachments
+}
+
+// renderBlocks renders a list of blocks at the given indentation depth.
+func renderBlocks(b *strings.Builder, blocks []notionBlock, depth int, attachments *[]attachment) {
+	inList := false
+
+	for i, block := range blocks {
+		isList := block.Type == "bulleted_list_item" || block.Type == "numbered_list_item" || block.Type == "to_do"
+
+		// End list spacing: if previous was list and current is not
+		if inList && !isList {
+			b.WriteString("\n")
+			inList = false
+		}
+
+		renderBlock(b, block, depth, attachments, i, blocks)
+
+		if isList {
+			inList = true
+		}
+	}
+
+	// Close trailing list
+	if inList {
+		b.WriteString("\n")
+	}
+}
+
+// renderBlock renders a single block to Markdown.
+func renderBlock(b *strings.Builder, block notionBlock, depth int, attachments *[]attachment, index int, siblings []notionBlock) {
+	indent := strings.Repeat("  ", depth)
+
+	switch block.Type {
+	case "paragraph":
+		rt := extractRichText(block.RawContent)
+		text := renderRichText(rt)
+		b.WriteString(indent + text + "\n\n")
+
+	case "heading_1", "heading_2", "heading_3", "heading_4":
+		level := int(block.Type[len(block.Type)-1] - '0')
+		rt := extractRichText(block.RawContent)
+		b.WriteString(strings.Repeat("#", level) + " " + renderRichText(rt) + "\n\n")
+		if len(block.Children) > 0 {
+			renderBlocks(b, block.Children, depth, attachments)
+		}
+
+	case "bulleted_list_item":
+		rt := extractRichText(block.RawContent)
+		b.WriteString(indent + "- " + renderRichText(rt) + "\n")
+		if len(block.Children) > 0 {
+			renderBlocks(b, block.Children, depth+1, attachments)
+		}
+
+	case "numbered_list_item":
+		rt := extractRichText(block.RawContent)
+		// Find position among consecutive numbered_list_items
+		num := 1
+		for j := index - 1; j >= 0 && siblings[j].Type == "numbered_list_item"; j-- {
+			num++
+		}
+		b.WriteString(fmt.Sprintf("%s%d. %s\n", indent, num, renderRichText(rt)))
+		if len(block.Children) > 0 {
+			renderBlocks(b, block.Children, depth+1, attachments)
+		}
+
+	case "to_do":
+		rt := extractRichText(block.RawContent)
+		checked := extractBool(block.RawContent, "checked")
+		checkbox := "[ ]"
+		if checked {
+			checkbox = "[x]"
+		}
+		b.WriteString(fmt.Sprintf("%s- %s %s\n", indent, checkbox, renderRichText(rt)))
+
+	case "toggle":
+		rt := extractRichText(block.RawContent)
+		b.WriteString(fmt.Sprintf("<details><summary>%s</summary>\n\n", renderRichText(rt)))
+		if len(block.Children) > 0 {
+			renderBlocks(b, block.Children, depth, attachments)
+		}
+		b.WriteString("</details>\n\n")
+
+	case "code":
+		rt := extractRichText(block.RawContent)
+		lang := extractString(block.RawContent, "language")
+		b.WriteString(fmt.Sprintf("```%s\n%s\n```\n\n", lang, renderRichText(rt)))
+
+	case "quote", "meeting_notes":
+		rt := extractRichText(block.RawContent)
+		text := renderRichText(rt)
+		for _, line := range strings.Split(text, "\n") {
+			b.WriteString("> " + line + "\n")
+		}
+		b.WriteString("\n")
+		if len(block.Children) > 0 {
+			renderBlocks(b, block.Children, depth, attachments)
+		}
+
+	case "callout":
+		rt := extractRichText(block.RawContent)
+		icon := extractIcon(block.RawContent)
+		prefix := ""
+		if icon != "" {
+			prefix = icon + " "
+		}
+		b.WriteString("> " + prefix + renderRichText(rt) + "\n\n")
+		if len(block.Children) > 0 {
+			renderBlocks(b, block.Children, depth, attachments)
+		}
+
+	case "divider":
+		b.WriteString("---\n\n")
+
+	case "equation":
+		expr := extractString(block.RawContent, "expression")
+		b.WriteString("$$" + expr + "$$\n\n")
+
+	case "table":
+		renderTable(b, block)
+
+	case "image":
+		file, caption := extractFileAndCaption(block.RawContent)
+		url := file.GetURL()
+		b.WriteString(fmt.Sprintf("![%s](%s)\n\n", caption, url))
+		if url != "" {
+			*attachments = append(*attachments, attachment{
+				URL:      url,
+				FileName: fileNameFromURL(url, "image"),
+				Type:     "image",
+			})
+		}
+
+	case "file", "pdf", "video", "audio":
+		renderMediaBlock(b, block, attachments)
+
+	case "bookmark", "link_preview":
+		url := extractString(block.RawContent, "url")
+		caption := ""
+		if rt := extractCaptionText(block.RawContent); rt != "" {
+			caption = rt
+		} else {
+			caption = url
+		}
+		b.WriteString(fmt.Sprintf("[%s](%s)\n\n", caption, url))
+
+	case "embed":
+		url := extractString(block.RawContent, "url")
+		b.WriteString(fmt.Sprintf("[%s](%s)\n\n", url, url))
+
+	case "link_to_page":
+		// Render as a link; the actual page title is not available in block data
+		pageID := extractLinkToPageID(block.RawContent)
+		b.WriteString(fmt.Sprintf("[Page %s](https://notion.so/%s)\n\n", pageID, strings.ReplaceAll(pageID, "-", "")))
+
+	case "synced_block":
+		// Render children (the synced content)
+		if len(block.Children) > 0 {
+			renderBlocks(b, block.Children, depth, attachments)
+		}
+
+	case "column_list":
+		// Render each column sequentially
+		if len(block.Children) > 0 {
+			for _, col := range block.Children {
+				if len(col.Children) > 0 {
+					renderBlocks(b, col.Children, depth, attachments)
+				}
+			}
+		}
+
+	case "column":
+		// Handled by column_list
+		if len(block.Children) > 0 {
+			renderBlocks(b, block.Children, depth, attachments)
+		}
+
+	case "tab_list":
+		// Render each tab's content sequentially (similar to column_list)
+		if len(block.Children) > 0 {
+			for _, tab := range block.Children {
+				if len(tab.Children) > 0 {
+					renderBlocks(b, tab.Children, depth, attachments)
+				}
+			}
+		}
+
+	case "tab":
+		// Render tab content sequentially
+		if len(block.Children) > 0 {
+			renderBlocks(b, block.Children, depth, attachments)
+		}
+
+	case "child_page":
+		// Render as a link in the parent page so it's not empty.
+		// The child page itself is fetched as a separate knowledge item by connector layer.
+		title := extractString(block.RawContent, "title")
+		if title == "" {
+			title = "Untitled"
+		}
+		b.WriteString(fmt.Sprintf("- [%s](https://notion.so/%s)\n", title, strings.ReplaceAll(block.ID, "-", "")))
+
+	case "child_database":
+		title := extractString(block.RawContent, "title")
+		if title == "" {
+			title = "Database"
+		}
+		b.WriteString(fmt.Sprintf("- [%s](https://notion.so/%s)\n", title, strings.ReplaceAll(block.ID, "-", "")))
+
+	case "table_of_contents", "breadcrumb", "template", "table_row", "unsupported":
+		// Skip — no meaningful content
+
+	default:
+		// Unknown block type — skip silently for forward compatibility
+	}
+}
+
+// renderMediaBlock renders a file/pdf/video/audio block as a Markdown link and collects the attachment.
+func renderMediaBlock(b *strings.Builder, block notionBlock, attachments *[]attachment) {
+	file, _ := extractFileAndCaption(block.RawContent)
+	url := file.GetURL()
+	name := file.Name
+	if name == "" {
+		name = fileNameFromURL(url, block.Type)
+	}
+	b.WriteString(fmt.Sprintf("[%s](%s)\n\n", name, url))
+	if url != "" {
+		*attachments = append(*attachments, attachment{
+			URL:      url,
+			FileName: name,
+			Type:     block.Type,
+		})
+	}
+}
+
+// renderTable converts a table block (with table_row children) to Markdown.
+func renderTable(b *strings.Builder, block notionBlock) {
+	if len(block.Children) == 0 {
+		return
+	}
+
+	for i, row := range block.Children {
+		cells := extractTableCells(row.RawContent)
+		var parts []string
+		for _, cell := range cells {
+			parts = append(parts, strings.ReplaceAll(renderRichText(cell), "|", "\\|"))
+		}
+		b.WriteString("| " + strings.Join(parts, " | ") + " |\n")
+
+		// Markdown tables always need a separator after the first row
+		if i == 0 {
+			var sep []string
+			for range parts {
+				sep = append(sep, "---")
+			}
+			b.WriteString("| " + strings.Join(sep, " | ") + " |\n")
+		}
+	}
+	b.WriteString("\n")
+}
+
+// --- Rich text rendering ---
+
+// renderRichText converts a slice of rich text elements to Markdown.
+func renderRichText(texts []notionRichText) string {
+	var b strings.Builder
+	for _, rt := range texts {
+		text := richTextToString(rt)
+		text = applyAnnotations(text, rt.Annotations)
+		if rt.Href != "" && !rt.Annotations.Code {
+			text = fmt.Sprintf("[%s](%s)", text, rt.Href)
+		}
+		b.WriteString(text)
+	}
+	return b.String()
+}
+
+// richTextToString extracts the base text from a rich text element.
+func richTextToString(rt notionRichText) string {
+	switch rt.Type {
+	case "text":
+		if rt.Text != nil {
+			return rt.Text.Content
+		}
+		return rt.PlainText
+
+	case "mention":
+		if rt.Mention == nil {
+			return rt.PlainText
+		}
+		switch rt.Mention.Type {
+		case "date":
+			if rt.Mention.Date != nil {
+				if rt.Mention.Date.End != "" {
+					return rt.Mention.Date.Start + " \u2192 " + rt.Mention.Date.End
+				}
+				return rt.Mention.Date.Start
+			}
+		case "page":
+			if rt.Mention.Page != nil {
+				return rt.PlainText // Notion fills PlainText with page title
+			}
+		case "database", "data_source":
+			if rt.Mention.Database != nil {
+				return rt.PlainText
+			}
+		case "link_preview":
+			if rt.Mention.LinkPreview != nil {
+				return rt.Mention.LinkPreview.URL
+			}
+		}
+		return rt.PlainText
+
+	case "equation":
+		if rt.Equation != nil {
+			return "$" + rt.Equation.Expression + "$"
+		}
+		return rt.PlainText
+
+	default:
+		return rt.PlainText
+	}
+}
+
+// applyAnnotations wraps text with Markdown formatting based on annotations.
+func applyAnnotations(text string, ann notionAnnotations) string {
+	if text == "" {
+		return text
+	}
+	if ann.Code {
+		text = "`" + text + "`"
+	}
+	if ann.Bold && ann.Italic {
+		text = "***" + text + "***"
+	} else if ann.Bold {
+		text = "**" + text + "**"
+	} else if ann.Italic {
+		text = "*" + text + "*"
+	}
+	if ann.Strikethrough {
+		text = "~~" + text + "~~"
+	}
+	if ann.Underline {
+		text = "<u>" + text + "</u>"
+	}
+	return text
+}
+
+// --- Raw content extraction helpers ---
+
+func extractRichText(raw json.RawMessage) []notionRichText {
+	if raw == nil {
+		return nil
+	}
+	var content struct {
+		RichText []notionRichText `json:"rich_text"`
+	}
+	json.Unmarshal(raw, &content)
+	return content.RichText
+}
+
+func extractString(raw json.RawMessage, key string) string {
+	if raw == nil {
+		return ""
+	}
+	var m map[string]json.RawMessage
+	if err := json.Unmarshal(raw, &m); err != nil {
+		return ""
+	}
+	val, ok := m[key]
+	if !ok {
+		return ""
+	}
+	var s string
+	json.Unmarshal(val, &s)
+	return s
+}
+
+func extractBool(raw json.RawMessage, key string) bool {
+	if raw == nil {
+		return false
+	}
+	var m map[string]json.RawMessage
+	if err := json.Unmarshal(raw, &m); err != nil {
+		return false
+	}
+	val, ok := m[key]
+	if !ok {
+		return false
+	}
+	var b bool
+	json.Unmarshal(val, &b)
+	return b
+}
+
+func extractIcon(raw json.RawMessage) string {
+	if raw == nil {
+		return ""
+	}
+	var content struct {
+		Icon *struct {
+			Type  string `json:"type"`
+			Emoji string `json:"emoji,omitempty"`
+		} `json:"icon"`
+	}
+	json.Unmarshal(raw, &content)
+	if content.Icon != nil && content.Icon.Type == "emoji" {
+		return content.Icon.Emoji
+	}
+	return ""
+}
+
+func extractFileAndCaption(raw json.RawMessage) (notionFile, string) {
+	var file notionFile
+	if raw != nil {
+		json.Unmarshal(raw, &file)
+	}
+	captionText := ""
+	if len(file.Caption) > 0 {
+		captionText = renderRichText(file.Caption)
+	}
+	return file, captionText
+}
+
+func extractCaptionText(raw json.RawMessage) string {
+	if raw == nil {
+		return ""
+	}
+	var content struct {
+		Caption []notionRichText `json:"caption"`
+	}
+	json.Unmarshal(raw, &content)
+	if len(content.Caption) > 0 {
+		return renderRichText(content.Caption)
+	}
+	return ""
+}
+
+func extractTableCells(raw json.RawMessage) [][]notionRichText {
+	if raw == nil {
+		return nil
+	}
+	var content struct {
+		Cells [][]notionRichText `json:"cells"`
+	}
+	json.Unmarshal(raw, &content)
+	return content.Cells
+}
+
+func extractLinkToPageID(raw json.RawMessage) string {
+	if raw == nil {
+		return ""
+	}
+	var content struct {
+		Type   string `json:"type"`
+		PageID string `json:"page_id,omitempty"`
+	}
+	json.Unmarshal(raw, &content)
+	return content.PageID
+}
+
+func fileNameFromURL(url, fallbackType string) string {
+	if url == "" {
+		return fallbackType
+	}
+	// Extract filename from URL path (before query params)
+	path := url
+	if idx := strings.Index(path, "?"); idx >= 0 {
+		path = path[:idx]
+	}
+	if idx := strings.LastIndex(path, "/"); idx >= 0 {
+		name := path[idx+1:]
+		if name != "" {
+			return name
+		}
+	}
+	return fallbackType
+}
@@ -0,0 +1,352 @@
+package notion
+
+import (
+	"encoding/json"
+	"testing"
+)
+
+func TestRenderRichText(t *testing.T) {
+	tests := []struct {
+		name     string
+		texts    []notionRichText
+		expected string
+	}{
+		{
+			name: "plain text",
+			texts: []notionRichText{
+				{Type: "text", PlainText: "hello", Text: &notionTextContent{Content: "hello"}, Annotations: notionAnnotations{}},
+			},
+			expected: "hello",
+		},
+		{
+			name: "bold text",
+			texts: []notionRichText{
+				{Type: "text", PlainText: "bold", Text: &notionTextContent{Content: "bold"}, Annotations: notionAnnotations{Bold: true}},
+			},
+			expected: "**bold**",
+		},
+		{
+			name: "italic text",
+			texts: []notionRichText{
+				{Type: "text", PlainText: "italic", Text: &notionTextContent{Content: "italic"}, Annotations: notionAnnotations{Italic: true}},
+			},
+			expected: "*italic*",
+		},
+		{
+			name: "code text",
+			texts: []notionRichText{
+				{Type: "text", PlainText: "code", Text: &notionTextContent{Content: "code"}, Annotations: notionAnnotations{Code: true}},
+			},
+			expected: "`code`",
+		},
+		{
+			name: "strikethrough",
+			texts: []notionRichText{
+				{Type: "text", PlainText: "del", Text: &notionTextContent{Content: "del"}, Annotations: notionAnnotations{Strikethrough: true}},
+			},
+			expected: "~~del~~",
+		},
+		{
+			name: "underline",
+			texts: []notionRichText{
+				{Type: "text", PlainText: "ul", Text: &notionTextContent{Content: "ul"}, Annotations: notionAnnotations{Underline: true}},
+			},
+			expected: "<u>ul</u>",
+		},
+		{
+			name: "bold italic combo",
+			texts: []notionRichText{
+				{Type: "text", PlainText: "bi", Text: &notionTextContent{Content: "bi"}, Annotations: notionAnnotations{Bold: true, Italic: true}},
+			},
+			expected: "***bi***",
+		},
+		{
+			name: "text with link",
+			texts: []notionRichText{
+				{Type: "text", PlainText: "click", Href: "https://example.com", Text: &notionTextContent{Content: "click"}, Annotations: notionAnnotations{}},
+			},
+			expected: "[click](https://example.com)",
+		},
+		{
+			name: "equation",
+			texts: []notionRichText{
+				{Type: "equation", PlainText: "E=mc^2", Equation: &notionEquation{Expression: "E=mc^2"}, Annotations: notionAnnotations{}},
+			},
+			expected: "$E=mc^2$",
+		},
+		{
+			name: "date mention",
+			texts: []notionRichText{
+				{Type: "mention", PlainText: "2026-01-15", Mention: &notionMention{
+					Type: "date", Date: &notionDateMention{Start: "2026-01-15"},
+				}, Annotations: notionAnnotations{}},
+			},
+			expected: "2026-01-15",
+		},
+		{
+			name: "date mention with end",
+			texts: []notionRichText{
+				{Type: "mention", PlainText: "range", Mention: &notionMention{
+					Type: "date", Date: &notionDateMention{Start: "2026-01-15", End: "2026-01-20"},
+				}, Annotations: notionAnnotations{}},
+			},
+			expected: "2026-01-15 \u2192 2026-01-20",
+		},
+		{
+			name: "multiple segments",
+			texts: []notionRichText{
+				{Type: "text", PlainText: "hello ", Text: &notionTextContent{Content: "hello "}, Annotations: notionAnnotations{}},
+				{Type: "text", PlainText: "world", Text: &notionTextContent{Content: "world"}, Annotations: notionAnnotations{Bold: true}},
+			},
+			expected: "hello **world**",
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			got := renderRichText(tt.texts)
+			if got != tt.expected {
+				t.Errorf("renderRichText() = %q, want %q", got, tt.expected)
+			}
+		})
+	}
+}
+
+func makeBlock(t *testing.T, blockType string, content interface{}) notionBlock {
+	t.Helper()
+	raw, err := json.Marshal(content)
+	if err != nil {
+		t.Fatalf("marshal block content: %v", err)
+	}
+	return notionBlock{
+		ID:         "blk-test",
+		Type:       blockType,
+		RawContent: raw,
+	}
+}
+
+func TestBlocksToMarkdown_Paragraph(t *testing.T) {
+	blocks := []notionBlock{
+		makeBlock(t, "paragraph", map[string]interface{}{
+			"rich_text": []notionRichText{
+				{Type: "text", PlainText: "Hello world", Text: &notionTextContent{Content: "Hello world"}, Annotations: notionAnnotations{}},
+			},
+		}),
+	}
+	md, attachments := BlocksToMarkdown(blocks)
+	if md != "Hello world\n\n" {
+		t.Errorf("markdown = %q", md)
+	}
+	if len(attachments) != 0 {
+		t.Errorf("expected 0 attachments, got %d", len(attachments))
+	}
+}
+
+func TestBlocksToMarkdown_Headings(t *testing.T) {
+	blocks := []notionBlock{
+		makeBlock(t, "heading_1", map[string]interface{}{
+			"rich_text": []notionRichText{
+				{Type: "text", PlainText: "H1", Text: &notionTextContent{Content: "H1"}, Annotations: notionAnnotations{}},
+			},
+		}),
+		makeBlock(t, "heading_2", map[string]interface{}{
+			"rich_text": []notionRichText{
+				{Type: "text", PlainText: "H2", Text: &notionTextContent{Content: "H2"}, Annotations: notionAnnotations{}},
+			},
+		}),
+		makeBlock(t, "heading_3", map[string]interface{}{
+			"rich_text": []notionRichText{
+				{Type: "text", PlainText: "H3", Text: &notionTextContent{Content: "H3"}, Annotations: notionAnnotations{}},
+			},
+		}),
+	}
+	md, _ := BlocksToMarkdown(blocks)
+	expected := "# H1\n\n## H2\n\n### H3\n\n"
+	if md != expected {
+		t.Errorf("markdown = %q, want %q", md, expected)
+	}
+}
+
+func TestBlocksToMarkdown_Code(t *testing.T) {
+	blocks := []notionBlock{
+		makeBlock(t, "code", map[string]interface{}{
+			"language": "go",
+			"rich_text": []notionRichText{
+				{Type: "text", PlainText: "fmt.Println()", Text: &notionTextContent{Content: "fmt.Println()"}, Annotations: notionAnnotations{}},
+			},
+		}),
+	}
+	md, _ := BlocksToMarkdown(blocks)
+	expected := "```go\nfmt.Println()\n```\n\n"
+	if md != expected {
+		t.Errorf("markdown = %q, want %q", md, expected)
+	}
+}
+
+func TestBlocksToMarkdown_List(t *testing.T) {
+	blocks := []notionBlock{
+		makeBlock(t, "bulleted_list_item", map[string]interface{}{
+			"rich_text": []notionRichText{
+				{Type: "text", PlainText: "Item 1", Text: &notionTextContent{Content: "Item 1"}, Annotations: notionAnnotations{}},
+			},
+		}),
+		makeBlock(t, "bulleted_list_item", map[string]interface{}{
+			"rich_text": []notionRichText{
+				{Type: "text", PlainText: "Item 2", Text: &notionTextContent{Content: "Item 2"}, Annotations: notionAnnotations{}},
+			},
+		}),
+	}
+	md, _ := BlocksToMarkdown(blocks)
+	expected := "- Item 1\n- Item 2\n\n"
+	if md != expected {
+		t.Errorf("markdown = %q, want %q", md, expected)
+	}
+}
+
+func TestBlocksToMarkdown_ToDo(t *testing.T) {
+	blocks := []notionBlock{
+		makeBlock(t, "to_do", map[string]interface{}{
+			"checked": false,
+			"rich_text": []notionRichText{
+				{Type: "text", PlainText: "Unchecked", Text: &notionTextContent{Content: "Unchecked"}, Annotations: notionAnnotations{}},
+			},
+		}),
+		makeBlock(t, "to_do", map[string]interface{}{
+			"checked": true,
+			"rich_text": []notionRichText{
+				{Type: "text", PlainText: "Checked", Text: &notionTextContent{Content: "Checked"}, Annotations: notionAnnotations{}},
+			},
+		}),
+	}
+	md, _ := BlocksToMarkdown(blocks)
+	expected := "- [ ] Unchecked\n- [x] Checked\n\n"
+	if md != expected {
+		t.Errorf("markdown = %q, want %q", md, expected)
+	}
+}
+
+func TestBlocksToMarkdown_Quote(t *testing.T) {
+	blocks := []notionBlock{
+		makeBlock(t, "quote", map[string]interface{}{
+			"rich_text": []notionRichText{
+				{Type: "text", PlainText: "A wise quote", Text: &notionTextContent{Content: "A wise quote"}, Annotations: notionAnnotations{}},
+			},
+		}),
+	}
+	md, _ := BlocksToMarkdown(blocks)
+	expected := "> A wise quote\n\n"
+	if md != expected {
+		t.Errorf("markdown = %q, want %q", md, expected)
+	}
+}
+
+func TestBlocksToMarkdown_Divider(t *testing.T) {
+	blocks := []notionBlock{
+		{ID: "blk-div", Type: "divider"},
+	}
+	md, _ := BlocksToMarkdown(blocks)
+	if md != "---\n\n" {
+		t.Errorf("markdown = %q", md)
+	}
+}
+
+func TestBlocksToMarkdown_Equation(t *testing.T) {
+	blocks := []notionBlock{
+		makeBlock(t, "equation", map[string]interface{}{
+			"expression": "E = mc^2",
+		}),
+	}
+	md, _ := BlocksToMarkdown(blocks)
+	expected := "$$E = mc^2$$\n\n"
+	if md != expected {
+		t.Errorf("markdown = %q, want %q", md, expected)
+	}
+}
+
+func TestBlocksToMarkdown_Image(t *testing.T) {
+	blocks := []notionBlock{
+		makeBlock(t, "image", map[string]interface{}{
+			"type": "file",
+			"file": map[string]interface{}{
+				"url":         "https://s3.example.com/img.png",
+				"expiry_time": "2026-01-15T11:00:00.000Z",
+			},
+			"caption": []notionRichText{
+				{Type: "text", PlainText: "A photo", Text: &notionTextContent{Content: "A photo"}, Annotations: notionAnnotations{}},
+			},
+		}),
+	}
+	md, attachments := BlocksToMarkdown(blocks)
+	if md != "![A photo](https://s3.example.com/img.png)\n\n" {
+		t.Errorf("markdown = %q", md)
+	}
+	if len(attachments) != 1 {
+		t.Fatalf("expected 1 attachment, got %d", len(attachments))
+	}
+	if attachments[0].URL != "https://s3.example.com/img.png" {
+		t.Errorf("attachment URL = %q", attachments[0].URL)
+	}
+	if attachments[0].Type != "image" {
+		t.Errorf("attachment Type = %q", attachments[0].Type)
+	}
+}
+
+func TestBlocksToMarkdown_UnknownType(t *testing.T) {
+	blocks := []notionBlock{
+		{ID: "blk-unknown", Type: "future_block_type"},
+	}
+	md, _ := BlocksToMarkdown(blocks)
+	if md != "" {
+		t.Errorf("unknown block should produce empty output, got %q", md)
+	}
+}
+
+func TestBlocksToMarkdown_Table(t *testing.T) {
+	tableBlock := notionBlock{
+		ID:          "blk-table",
+		Type:        "table",
+		HasChildren: true,
+		RawContent:  json.RawMessage(`{"table_width": 2, "has_column_header": true}`),
+		Children: []notionBlock{
+			makeBlock(t, "table_row", map[string]interface{}{
+				"cells": [][]notionRichText{
+					{{Type: "text", PlainText: "Name", Text: &notionTextContent{Content: "Name"}, Annotations: notionAnnotations{}}},
+					{{Type: "text", PlainText: "Age", Text: &notionTextContent{Content: "Age"}, Annotations: notionAnnotations{}}},
+				},
+			}),
+			makeBlock(t, "table_row", map[string]interface{}{
+				"cells": [][]notionRichText{
+					{{Type: "text", PlainText: "Alice", Text: &notionTextContent{Content: "Alice"}, Annotations: notionAnnotations{}}},
+					{{Type: "text", PlainText: "30", Text: &notionTextContent{Content: "30"}, Annotations: notionAnnotations{}}},
+				},
+			}),
+		},
+	}
+	md, _ := BlocksToMarkdown([]notionBlock{tableBlock})
+	expected := "| Name | Age |\n| --- | --- |\n| Alice | 30 |\n\n"
+	if md != expected {
+		t.Errorf("markdown = %q, want %q", md, expected)
+	}
+}
+
+func TestBlocksToMarkdown_NestedList(t *testing.T) {
+	parent := makeBlock(t, "bulleted_list_item", map[string]interface{}{
+		"rich_text": []notionRichText{
+			{Type: "text", PlainText: "Parent", Text: &notionTextContent{Content: "Parent"}, Annotations: notionAnnotations{}},
+		},
+	})
+	parent.HasChildren = true
+	parent.Children = []notionBlock{
+		makeBlock(t, "bulleted_list_item", map[string]interface{}{
+			"rich_text": []notionRichText{
+				{Type: "text", PlainText: "Child", Text: &notionTextContent{Content: "Child"}, Annotations: notionAnnotations{}},
+			},
+		}),
+	}
+
+	md, _ := BlocksToMarkdown([]notionBlock{parent})
+	expected := "- Parent\n  - Child\n\n"
+	if md != expected {
+		t.Errorf("markdown = %q, want %q", md, expected)
+	}
+}