Rakka/Storage/Impl.hs

   1 module Rakka.Storage.Impl
   2     ( getPage'
   3     , putPage'
   4     , startIndexManager
   5     )
   6     where
   7
   8 import           Control.Concurrent
   9 import           Control.Concurrent.STM
  10 import           Control.Exception
  11 import           Control.Monad
  12 import           Data.Maybe
  13 import           Data.Set (Set)
  14 import qualified Data.Set as S
  15 import           Network.HTTP.Lucu
  16 import           Network.URI
  17 import           Rakka.Page
  18 import           Rakka.Storage.DefaultPage
  19 import           Rakka.Storage.Repos
  20 import           Rakka.Storage.Types
  21 import           Subversion.Types
  22 import           Subversion.FileSystem
  23 import           Subversion.Repository
  24 import           System.Directory
  25 import           System.FilePath
  26 import           System.IO
  27 import           System.Log.Logger
  28 import           System.Posix.Files
  29 import           System.Posix.Types
  30 import           System.Posix.IO
  31 import           Text.HyperEstraier hiding (WriteLock)
  32
  33
  34 logger :: String
  35 logger = "Rakka.Storage"
  36
  37
  38 getPage' :: Repository -> PageName -> Maybe RevNum -> IO (Maybe Page)
  39 getPage' repos name rev
  40     = do page <- loadPageInRepository repos name rev
  41          case page of
  42            Nothing -> loadDefaultPage name
  43            p       -> return p
  44
  45
  46 putPage' :: Repository -> Page -> IO StatusCode
  47 putPage' = putPageIntoRepository
  48
  49
  50 findAllPages :: Repository -> RevNum -> IO (Set PageName)
  51 findAllPages _     0   = findAllDefaultPages
  52 findAllPages repos rev = do reposPages   <- findAllPagesInRevision repos rev
  53                             defaultPages <- findAllDefaultPages
  54                             return (reposPages `S.union` defaultPages)
  55
  56
  57 findChangedPages :: Repository -> RevNum -> RevNum -> IO (Set PageName)
  58 findChangedPages repos 0      newRev = findAllPages repos newRev
  59 findChangedPages repos oldRev newRev
  60     = mapM (findChangedPagesAtRevision repos) [oldRev + 1 .. newRev]
  61       >>=
  62       return . S.unions
  63
  64
  65 getCurrentRevNum :: Repository -> IO RevNum
  66 getCurrentRevNum repos
  67     = getRepositoryFS repos >>= getYoungestRev
  68
  69
  70 startIndexManager :: FilePath -> Repository -> (Page -> IO Document) -> IO (TChan IndexReq)
  71 startIndexManager lsdir repos mkDraft
  72     = do chan  <- newTChanIO
  73          index <- openIndex indexDir revFile
  74          forkIO (loop chan index)
  75          return chan
  76     where
  77       indexDir = lsdir </> "index"
  78       revFile  = lsdir </> "indexRev"
  79
  80       loop :: TChan IndexReq -> Database -> IO ()
  81       loop chan index
  82           = do req <- atomically $ readTChan chan
  83                case req of
  84                  RebuildIndex
  85                      -> do noticeM logger "Rebuilding the H.E. index..."
  86                            closeDatabase index
  87                            removeDirectoryRecursive indexDir
  88                            index' <- openIndex indexDir revFile
  89                            syncIndex' index' revFile repos mkDraft
  90                            loop chan index'
  91
  92                  SyncIndex
  93                      -> do syncIndex' index revFile repos mkDraft
  94                            loop chan index
  95
  96                  SearchIndex cond var
  97                      -> do result <- searchIndex index cond
  98                            atomically $ putTMVar var result
  99                            loop chan index
 100
 101
 102 -- casket を R/W モードで開く。成功したらそのまま返し、失敗したら
 103 -- indexDir と revFile を削除してから casket を R/W モードで開く。
 104 openIndex :: FilePath -> FilePath -> IO Database
 105 openIndex indexDir revFile
 106     = do ret <- openDatabase indexDir (Writer [])
 107          case ret of
 108            Right index
 109                -> do debugM logger ("Opened an H.E. index on " ++ indexDir)
 110                      return index
 111
 112            Left err
 113                -> do noticeM logger ("Failed to open an H.E. index on "
 114                                      ++ indexDir ++ ": " ++ show err)
 115
 116                      indexExists <- doesDirectoryExist indexDir
 117                      when indexExists
 118                               $ removeDirectoryRecursive indexDir
 119
 120                      revFileExists <- doesFileExist revFile
 121                      when revFileExists
 122                               $ removeFile revFile
 123
 124                      Right index <- openDatabase indexDir (Writer [Create []])
 125                      addAttrIndex index "@uri"           SeqIndex
 126                      addAttrIndex index "rakka:revision" SeqIndex
 127                      noticeM logger ("Created an H.E. index on " ++ indexDir)
 128
 129                      return index
 130
 131
 132 syncIndex' :: Database -> FilePath -> Repository -> (Page -> IO Document) -> IO ()
 133 syncIndex' index revFile repos mkDraft
 134     = updateIndexRev revFile $ \ oldRev ->
 135       do debugM logger ("The index revision is currently " ++ show oldRev)
 136
 137          newRev <- getCurrentRevNum repos
 138          debugM logger ("The repository revision is currently " ++ show newRev)
 139
 140          when (oldRev == 0 || newRev /= oldRev)
 141               $ syncIndex'' oldRev newRev
 142          return newRev
 143     where
 144       syncIndex'' :: RevNum -> RevNum -> IO ()
 145       syncIndex'' oldRev newRev
 146           = do pages <- findChangedPages repos oldRev newRev
 147                mapM_ (updateIndex index repos mkDraft newRev) (S.toList pages)
 148
 149
 150 searchIndex :: Database -> Condition -> IO [(PageName, RevNum)]
 151 searchIndex index cond
 152     = searchDatabase index cond >>= mapM fromId
 153     where
 154       fromId :: DocumentID -> IO (PageName, RevNum)
 155       fromId docId
 156           = do uri <- getDocURI index docId
 157                rev <- getDocAttr index docId "rakka:revision"
 158                       >>= return . read . fromJust
 159                return (decodePageName $ uriPath uri, rev)
 160
 161
 162 updateIndex :: Database
 163             -> Repository
 164             -> (Page -> IO Document)
 165             -> RevNum
 166             -> PageName
 167             -> IO ()
 168 updateIndex index repos mkDraft rev name
 169     = do pageM <- getPage' repos name (Just rev)
 170          case pageM of
 171            -- ページが削除された
 172            Nothing
 173                -> do docIdM <- getDocIdByURI index (mkRakkaURI name)
 174                      case docIdM of
 175                        Nothing    -> return ()
 176                        Just docId -> do removeDocument index docId [CleaningRemove]
 177                                         infoM logger ("Removed page " ++ name ++ " from the index")
 178            Just page
 179                -> do draft <- mkDraft page
 180                      putDocument index draft [CleaningPut]
 181                      infoM logger ("Indexed page " ++ name ++ " of revision " ++ show (pageRevision page))
 182
 183
 184 updateIndexRev :: FilePath -> (RevNum -> IO RevNum) -> IO ()
 185 updateIndexRev revFile f = bracket acquireLock releaseLock update
 186     where
 187       acquireLock :: IO Fd
 188       acquireLock
 189           = do fd <- openFd revFile ReadWrite (Just stdFileMode) defaultFileFlags
 190                waitToSetLock fd (WriteLock, AbsoluteSeek, 0, 0)
 191                return fd
 192
 193       releaseLock :: Fd -> IO ()
 194       releaseLock fd
 195           = setLock fd (Unlock, AbsoluteSeek, 0, 0)
 196
 197       update :: Fd -> IO ()
 198       update fd
 199           = do fdSeek fd AbsoluteSeek 0
 200                size <- return . fromIntegral . fileSize =<< getFdStatus fd
 201                (revStr, gotSize) <- fdRead fd size
 202                when (size /= gotSize) $ fail ("read " ++ show gotSize ++
 203                                               " bytes but expected " ++ show size ++ " bytes")
 204
 205                let rev = case revStr of
 206                            "" -> 0
 207                            _  -> read revStr
 208
 209                rev' <- f rev
 210
 211                let revStr' = show rev' ++ "\n"
 212                    size'   = fromIntegral $ length revStr'
 213
 214                fdSeek fd AbsoluteSeek 0
 215                setFdSize fd 0
 216                wroteSize <- fdWrite fd revStr'
 217                when (size' /= wroteSize) $ fail ("wrote " ++ show wroteSize ++
 218                                                  " bytes but expected " ++ show size' ++ " bytes")